VRIO/src/VRio.Core/Device/VRioPipeService.cs

using System.Diagnostics;
using System.IO.Pipes;
using System.Runtime.InteropServices;

namespace VRio.Core.Device;

/// <summary>
/// Serves a <see cref="VRioDevice"/> over the named pipe that the DOSBox-X
/// fork's <c>serial1=namedpipe pipe:vrio</c> backend connects to — the
/// com0com-free transport. vRIO is the pipe <em>server</em> (the device is
/// always present); DOSBox-X is the client and background-retries while the
/// pipe is missing, so this service listens for the whole app lifetime
/// alongside the COM path at no cost. An unconnected pipe is an unplugged
/// cable: the device keeps running, transmit bytes fall on the floor.
///
/// <para>Framing per <see cref="PipeFraming"/>. On connect we immediately
/// send our lines frame (DTR+RTS high — "board present", the same lines the
/// serial path asserts via DtrEnable/RtsEnable); the peer's DTR edges surface
/// through <see cref="HostHandshake"/> exactly like the DSR blips through a
/// null modem, and a disconnect drops all lines low. The game hardware-resets
/// the RIO by pulsing DTR, and the in-band lines frames keep that edge's
/// position in the byte stream — the reason the contract multiplexes control
/// onto the data pipe instead of using a second one.</para>
///
/// <para>Outbound bytes are paced one per 10-bit frame time exactly like
/// <see cref="VRioSerialService"/> (see its remarks for the full rationale).
/// With the 9600-baud wire gone, this pacer is the only thing standing
/// between the host and an impossible burst, so each paced byte travels as
/// its own 3-byte data frame and the inter-byte timing survives the pipe.</para>
/// </summary>
public sealed class VRioPipeService : IDisposable
{
    /// <summary>The contract pipe name: DOSBox-X connects to <c>\\.\pipe\vrio</c>.</summary>
    public const string DefaultPipeName = "vrio";

    // Pacing mirrors VRioSerialService: one byte per 10-bit frame at 9600
    // baud; sleep until ~1.8 ms from the slot, then spin the remainder.
    private static readonly long BytePeriodTicks = Stopwatch.Frequency * 10 / VRioSerialService.BaudRate;
    private static readonly long SpinThresholdTicks = Stopwatch.Frequency * 18 / 10_000;

    private readonly VRioDevice _device;
    private readonly string _pipeName;
    private readonly object _txGate = new();
    private readonly Queue<byte> _txQueue = new();
    private readonly PipeFrameDecoder _decoder = new();

    private NamedPipeServerStream? _pipe;
    private Thread? _server;
    private Thread? _writer;
    private volatile bool _running;
    private volatile bool _clientConnected;
    private bool _peerDtr, _peerRts;
    private bool _timerResolutionRaised;

    public VRioPipeService(VRioDevice device, string pipeName = DefaultPipeName)
    {
        _device = device ?? throw new ArgumentNullException(nameof(device));
        if (string.IsNullOrWhiteSpace(pipeName))
            throw new ArgumentException("Pipe name is required.", nameof(pipeName));
        _pipeName = pipeName;

        _device.Transmit += Write;
        _decoder.Data += (buffer, count) => _device.OnReceived(buffer, count);
        _decoder.Lines += OnPeerLines;
    }

    /// <summary>The served pipe name (without the <c>\\.\pipe\</c> prefix).</summary>
    public string PipeName => _pipeName;

    /// <summary>True while a client is connected.</summary>
    public bool IsClientConnected => _clientConnected;

    /// <summary>A client connected (true) or went away (false).</summary>
    public event Action<bool>? ClientChanged;

    /// <summary>
    /// The host's DTR line changed (the game pulses it to hardware-reset the
    /// board). Same semantics as <see cref="VRioSerialService.HostHandshake"/>:
    /// the argument is the new line state as seen on our DSR pin.
    /// </summary>
    public event Action<bool>? HostHandshake;

    /// <summary>Pipe-level log lines (listen/connect/errors).</summary>
    public event Action<string>? Logged;

    /// <summary>Start listening (idempotent). Clients may come and go forever.</summary>
    public void Start()
    {
        if (_running)
            return;

        _running = true;
        lock (_txGate) _txQueue.Clear();

        _server = new Thread(ServerLoop) { IsBackground = true, Name = $"vRIO pipe server ({_pipeName})" };
        _server.Start();
        _writer = new Thread(WriteLoop) { IsBackground = true, Name = $"vRIO pipe writer ({_pipeName})" };
        _writer.Start();

        Logged?.Invoke($@"Listening on \\.\pipe\{_pipeName} (TX paced at the wire rate) — DOSBox-X connects when it boots");
    }

    /// <summary>Stop listening and drop any client (idempotent).</summary>
    public void Stop()
    {
        if (!_running)
            return;

        _running = false;
        lock (_txGate)
        {
            _txQueue.Clear();
            Monitor.PulseAll(_txGate); // wake the writer so it can exit
        }

        NamedPipeServerStream? pipe = _pipe;
        _pipe = null;
        try { pipe?.Dispose(); }
        catch (IOException) { }

        // A WaitForConnection pending on the disposed stream can survive the
        // Dispose on net48; a throwaway client connect releases it either way.
        try
        {
            using var poke = new NamedPipeClientStream(".", _pipeName, PipeDirection.Out);
            poke.Connect(100);
        }
        catch (Exception ex) when (ex is IOException or TimeoutException or UnauthorizedAccessException) { }

        _server?.Join(1000);
        _server = null;
        _writer?.Join(1000);
        _writer = null;

        if (_timerResolutionRaised)
        {
            timeEndPeriod(1);
            _timerResolutionRaised = false;
        }
        _clientConnected = false;

        Logged?.Invoke("Pipe server stopped");
    }

    private void ServerLoop()
    {
        bool busyLogged = false; // log the name collision once, not per retry

        while (_running)
        {
            NamedPipeServerStream pipe;
            try
            {
                pipe = new NamedPipeServerStream(_pipeName, PipeDirection.InOut, 1,
                    PipeTransmissionMode.Byte, PipeOptions.Asynchronous);
            }
            catch (Exception ex) when (ex is IOException or UnauthorizedAccessException)
            {
                // Name already served — most likely a second vRIO instance.
                if (!busyLogged)
                {
                    busyLogged = true;
                    Logged?.Invoke($@"\\.\pipe\{_pipeName} is busy ({ex.Message.TrimEnd('.')}) — retrying until it frees up");
                }
                for (int i = 0; i < 20 && _running; i++)
                    Thread.Sleep(100);
                continue;
            }
            busyLogged = false;

            _pipe = pipe;
            try
            {
                pipe.WaitForConnection();
            }
            catch (Exception ex) when (ex is IOException or ObjectDisposedException or InvalidOperationException)
            {
                // Disposed by Stop(), or a client vanished mid-connect.
                _pipe = null;
                try { pipe.Dispose(); }
                catch (IOException) { }
                continue;
            }

            if (!_running || !OnClientConnected(pipe))
            {
                _pipe = null;
                try { pipe.Dispose(); }
                catch (IOException) { }
                continue;
            }

            ReadUntilDisconnect(pipe);

            OnClientDisconnected();
            _pipe = null;
            try { pipe.Dispose(); }
            catch (IOException) { }
        }
    }

    /// <summary>Per-connection setup; false if the client died before it finished.</summary>
    private bool OnClientConnected(NamedPipeServerStream pipe)
    {
        _decoder.Reset();
        _peerDtr = _peerRts = false;
        lock (_txGate) _txQueue.Clear();

        // Our lines frame goes first, before the writer can interleave data
        // (it only writes once _clientConnected flips below): DTR+RTS high,
        // board present — the peer maps them to its DSR/CTS.
        try
        {
            byte[] lines = PipeFraming.EncodeLines(PipeFraming.LinesPresent);
            pipe.Write(lines, 0, lines.Length);
            pipe.Flush();
        }
        catch (Exception ex) when (ex is IOException or ObjectDisposedException or InvalidOperationException)
        {
            return false;
        }

        // 1 ms system timer resolution while a client is connected, so the
        // pacer's Thread.Sleep(1) actually sleeps ~1 ms.
        _timerResolutionRaised = timeBeginPeriod(1) == 0;
        _clientConnected = true;

        Logged?.Invoke("Pipe client connected — sent lines DTR+RTS (board present)");
        ClientChanged?.Invoke(true);
        return true;
    }

    private void OnClientDisconnected()
    {
        _clientConnected = false;
        lock (_txGate) _txQueue.Clear();

        if (_timerResolutionRaised)
        {
            timeEndPeriod(1);
            _timerResolutionRaised = false;
        }

        // Contract: a disconnect drops all lines low on the surviving side.
        if (_peerDtr)
        {
            _peerDtr = false;
            HostHandshake?.Invoke(false);
        }
        _peerRts = false;

        if (_running)
        {
            Logged?.Invoke("Pipe client disconnected");
            ClientChanged?.Invoke(false);
        }
    }

    private void ReadUntilDisconnect(NamedPipeServerStream pipe)
    {
        var buffer = new byte[512];
        while (_running)
        {
            int n;
            try
            {
                n = pipe.Read(buffer, 0, buffer.Length);
            }
            catch (Exception ex) when (ex is IOException or ObjectDisposedException or InvalidOperationException or OperationCanceledException)
            {
                if (_running)
                    Logged?.Invoke($"Pipe error: {ex.Message}");
                return;
            }

            if (n == 0)
                return; // client closed its end

            if (!_decoder.Feed(buffer, n))
            {
                Logged?.Invoke($"Pipe protocol violation: {_decoder.Violation} — dropping the connection");
                return;
            }
        }
    }

    private void OnPeerLines(byte lines)
    {
        _peerRts = (lines & PipeFraming.LineRts) != 0; // tracked; nothing consumes CTS today
        bool dtr = (lines & PipeFraming.LineDtr) != 0;
        if (dtr == _peerDtr)
            return;

        _peerDtr = dtr;
        HostHandshake?.Invoke(dtr);
    }

    // The device's Transmit handler: queue the frame for the paced writer so
    // the caller (UI click, reader thread mid-reply) never blocks on the pipe.
    private void Write(byte[] data)
    {
        if (!_running || !_clientConnected)
            return; // no host on the pipe — a real UART shifts into an unterminated line

        lock (_txGate)
        {
            foreach (byte b in data)
                _txQueue.Enqueue(b);
            Monitor.Pulse(_txGate);
        }
    }

    private void WriteLoop()
    {
        // One data frame per paced byte, so the pipe carries the same
        // inter-byte spacing a 9600-baud wire would.
        var frame = new byte[] { PipeFraming.DataType, 1, 0 };
        long slot = Stopwatch.GetTimestamp();

        while (_running)
        {
            lock (_txGate)
            {
                while (_txQueue.Count == 0)
                {
                    if (!_running)
                        return;
                    Monitor.Wait(_txGate, 200); // timed, so a missed pulse can't wedge shutdown
                }
                frame[2] = _txQueue.Dequeue();
            }

            // This byte's wire slot: one frame after the previous byte, or now
            // if the line has been idle (no burst "catch-up" debt).
            slot = Math.Max(slot + BytePeriodTicks, Stopwatch.GetTimestamp());
            PaceUntil(slot);

            NamedPipeServerStream? pipe = _pipe;
            if (pipe is null || !_clientConnected)
                continue; // client left while this byte waited its slot — it's stale, drop it

            try
            {
                pipe.Write(frame, 0, frame.Length);
                pipe.Flush();
            }
            catch (Exception ex) when (ex is IOException or ObjectDisposedException or InvalidOperationException)
            {
                // The client died mid-write; the backlog is already stale.
                // Drop it and keep serving — the server loop notices the EOF
                // and cycles back to WaitForConnection.
                lock (_txGate) _txQueue.Clear();
                continue;
            }

            // If the wait overshot its slot, pace the next byte from the
            // actual emission instead — a stall must not cause a catch-up burst.
            long now = Stopwatch.GetTimestamp();
            if (now > slot)
                slot = now;
        }
    }

    private static void PaceUntil(long slotTicks)
    {
        while (true)
        {
            long remaining = slotTicks - Stopwatch.GetTimestamp();
            if (remaining <= 0)
                return;
            if (remaining > SpinThresholdTicks)
                Thread.Sleep(1);
            else
                Thread.SpinWait(64);
        }
    }

    [DllImport("winmm.dll")]
    private static extern uint timeBeginPeriod(uint uMilliseconds);

    [DllImport("winmm.dll")]
    private static extern uint timeEndPeriod(uint uMilliseconds);

    public void Dispose()
    {
        _device.Transmit -= Write;
        Stop();
    }
}