merge: Task 5 — EtherNet/IP encapsulation codec + single-flight client

2026-10-07 16:14:02 -04:00
parent 7c81844be5 4c1c362b0e
commit 2f9c7eff58
6 changed files with 1133 additions and 0 deletions
@@ -0,0 +1,289 @@
+using System.Net.Sockets;
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip;
+
+/// <summary>
+///     One EtherNet/IP TCP session: connects, registers, and sends SendRRData (unconnected) and
+///     SendUnitData (connected) requests with <b>at most one request in flight</b>.
+/// </summary>
+/// <remarks>
+///     <para><b>Single flight.</b> Every exchange (write one frame, read until its reply) runs under a
+///     <see cref="SemaphoreSlim"/>(1,1). The connected sequence count is advanced inside that gate, so
+///     concurrent callers can never put sequences on the wire out of order.</para>
+///     <para><b>Caller cancellation never faults the connection, and never misattributes a reply</b>
+///     (rulings R5-1/R5-2). The caller gets <see cref="OperationCanceledException"/> at once, but the
+///     abandoned exchange keeps reading <i>its own</i> reply in the background while still holding the
+///     gate, bounded by the original timeout. The reply is discarded and the connection stays healthy;
+///     if the timeout elapses first the connection faults exactly as a normal timeout does. Because the
+///     next request is not written until the previous reply has been consumed, a late reply can never be
+///     taken for the next request's — which matters because ab_server zeroes the reply sender context,
+///     so context matching alone cannot tell two unconnected replies apart.</para>
+///     <para><b>Faults.</b> A timeout, an I/O error or an encapsulation-level protocol error
+///     (<see cref="AbStatus.BadReply"/>) closes the socket; <see cref="IsConnected"/> turns false and
+///     every later call throws <see cref="AbTransportException"/>(<see cref="AbStatus.BadConnection"/>).
+///     The owner reconnects with a fresh client.</para>
+///     <para><b>No half-written frames.</b> A frame write never observes a cancellation token.</para>
+/// </remarks>
+public sealed class EncapsulationClient : IAsyncDisposable
+{
+    private static readonly TimeSpan UnregisterBudget = TimeSpan.FromMilliseconds(500);
+
+    private readonly TcpClient _tcp;
+    private readonly NetworkStream _stream;
+
+    // Never disposed: an abandoned exchange may release it after DisposeAsync returns, and
+    // SemaphoreSlim.Release on a disposed instance throws. It holds no unmanaged resource unless
+    // AvailableWaitHandle is touched, which this class never does.
+    private readonly SemaphoreSlim _gate = new(1, 1);
+    private long _ctx;
+    private ushort _seq;             // connected sequence count; only touched under _gate
+    private volatile bool _faulted;
+    private int _disposed;
+
+    /// <summary>The session handle the target assigned in its RegisterSession reply.</summary>
+    public uint SessionHandle { get; private set; }
+
+    /// <summary>False once the connection has faulted or been disposed.</summary>
+    public bool IsConnected => !_faulted && Volatile.Read(ref _disposed) == 0 && _tcp.Connected;
+
+    private EncapsulationClient(TcpClient tcp)
+    {
+        _tcp = tcp;
+        _stream = tcp.GetStream();
+    }
+
+    /// <summary>Open the TCP connection and register an encapsulation session. <paramref name="timeout"/>
+    /// bounds the TCP connect and the RegisterSession exchange separately.</summary>
+    public static async Task<EncapsulationClient> ConnectAsync(string host, int port, TimeSpan timeout, CancellationToken ct)
+    {
+        var tcp = new TcpClient { NoDelay = true };
+        try
+        {
+            using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
+            cts.CancelAfter(timeout);
+            await tcp.ConnectAsync(host, port, cts.Token).ConfigureAwait(false);
+        }
+        catch (OperationCanceledException) when (!ct.IsCancellationRequested)
+        {
+            tcp.Dispose();
+            throw new AbTransportException(AbStatus.Timeout, $"TCP connect to {host}:{port} timed out.");
+        }
+        catch (SocketException ex)
+        {
+            tcp.Dispose();
+            throw new AbTransportException(AbStatus.BadConnection, $"TCP connect to {host}:{port} failed: {ex.SocketErrorCode}", ex);
+        }
+        catch
+        {
+            tcp.Dispose();
+            throw;
+        }
+
+        var client = new EncapsulationClient(tcp);
+        try
+        {
+            var (header, _) = await client.ExchangeAsync(connected: false,
+                (ctx, _) => EncapsulationCodec.RegisterSession(ctx), timeout, ct).ConfigureAwait(false);
+            client.SessionHandle = header.SessionHandle;
+            return client;
+        }
+        catch
+        {
+            client.Abort();
+            throw;
+        }
+    }
+
+    /// <summary>Send one unconnected (UCMM) CIP request and return the CIP reply bytes.</summary>
+    public async Task<ReadOnlyMemory<byte>> SendRRDataAsync(ReadOnlyMemory<byte> cip, TimeSpan timeout, CancellationToken ct)
+    {
+        var secs = (ushort)Math.Clamp((int)Math.Ceiling(timeout.TotalSeconds), 1, ushort.MaxValue);
+        var (_, reply) = await ExchangeAsync(connected: false,
+            (ctx, _) => EncapsulationCodec.SendRRData(SessionHandle, ctx, cip.Span, secs), timeout, ct).ConfigureAwait(false);
+        return reply;
+    }
+
+    /// <summary>Send one connected (class 3) CIP request over <paramref name="otConnectionId"/> and return
+    /// the CIP reply bytes. The client owns the connected sequence count (review B1).</summary>
+    public async Task<ReadOnlyMemory<byte>> SendUnitDataAsync(uint otConnectionId, ReadOnlyMemory<byte> cip,
+        TimeSpan timeout, CancellationToken ct)
+    {
+        var (_, reply) = await ExchangeAsync(connected: true,
+            (ctx, seq) => EncapsulationCodec.SendUnitData(SessionHandle, ctx, otConnectionId, seq, cip.Span), timeout, ct)
+            .ConfigureAwait(false);
+        return reply;
+    }
+
+    /// <summary>The connected sequence count after <paramref name="current"/>: increments, wrapping
+    /// 65535 → 1 (0 is never put on the wire).</summary>
+    internal static ushort NextSequence(ushort current) => current == ushort.MaxValue ? (ushort)1 : (ushort)(current + 1);
+
+    /// <summary>Acquire the gate, build the request (sequence assigned under the gate), then hand the
+    /// gate to <see cref="RunExchangeAsync"/>, which owns its release from that point on.</summary>
+    private async Task<(EncapsulationHeader Header, ReadOnlyMemory<byte> Cip)> ExchangeAsync(
+        bool connected, Func<ulong, ushort, byte[]> build, TimeSpan timeout, CancellationToken ct)
+    {
+        ThrowIfUnusable();
+        await _gate.WaitAsync(ct).ConfigureAwait(false);
+
+        Task<(EncapsulationHeader, ReadOnlyMemory<byte>)> exchange;
+        try
+        {
+            ThrowIfUnusable();                     // a drain we queued behind may have faulted
+            ct.ThrowIfCancellationRequested();     // nothing written yet: cancel cleanly
+            var ctx = (ulong)Interlocked.Increment(ref _ctx);
+            ushort seq = 0;
+            if (connected) seq = _seq = NextSequence(_seq);
+            var request = build(ctx, seq);
+            exchange = RunExchangeAsync(request, connected, ctx, seq, timeout);
+        }
+        catch
+        {
+            _gate.Release();                       // RunExchangeAsync never started: release here
+            throw;
+        }
+
+        try
+        {
+            return await exchange.WaitAsync(ct).ConfigureAwait(false);
+        }
+        catch (OperationCanceledException) when (ct.IsCancellationRequested)
+        {
+            // Caller abandoned the wait (R5-1). The exchange keeps draining its own reply under the
+            // gate and faults on its own timeout; observe its outcome so a fault is never unobserved.
+            _ = exchange.ContinueWith(static t => _ = t.Exception, CancellationToken.None,
+                TaskContinuationOptions.OnlyOnFaulted | TaskContinuationOptions.ExecuteSynchronously, TaskScheduler.Default);
+            throw;
+        }
+    }
+
+    /// <summary>Write one frame, then read frames until the matching reply arrives. Releases the gate
+    /// exactly once, in its <c>finally</c>. Never throws <see cref="OperationCanceledException"/>: every
+    /// failure surfaces as <see cref="AbTransportException"/> and faults the connection first.</summary>
+    private async Task<(EncapsulationHeader, ReadOnlyMemory<byte>)> RunExchangeAsync(
+        byte[] request, bool connected, ulong ctx, ushort seq, TimeSpan timeout)
+    {
+        var cmd = (ushort)(request[0] | (request[1] << 8));
+        try
+        {
+            using var timeoutCts = new CancellationTokenSource(timeout);
+            try
+            {
+                await _stream.WriteAsync(request, CancellationToken.None).ConfigureAwait(false);
+                while (true)
+                {
+                    var frame = await ReadFrameAsync(timeoutCts.Token).ConfigureAwait(false);
+                    if (TryMatch(frame, cmd, connected, ctx, seq, out var reply)) return reply;
+                }
+            }
+            catch (OperationCanceledException) when (timeoutCts.IsCancellationRequested)
+            {
+                Fault();
+                throw new AbTransportException(AbStatus.Timeout,
+                    $"No reply to command 0x{cmd:X4} within {timeout.TotalMilliseconds:F0} ms.");
+            }
+            catch (AbTransportException)
+            {
+                Fault();
+                throw;
+            }
+            catch (Exception ex) when (ex is IOException or SocketException or ObjectDisposedException
+                                           or EndOfStreamException or InvalidOperationException)
+            {
+                Fault();
+                throw new AbTransportException(AbStatus.BadConnection, $"Connection lost: {ex.Message}", ex);
+            }
+        }
+        finally
+        {
+            _gate.Release();
+        }
+    }
+
+    /// <summary>Does <paramref name="frame"/> answer the request in flight? Unconnected replies match on
+    /// command and sender context, tolerating a zeroed context (ab_server, R5-2); connected replies match
+    /// on the echoed sequence count. Anything else is skipped. A matching frame is fully validated, so an
+    /// encapsulation error throws BadReply.</summary>
+    private static bool TryMatch(byte[] frame, ushort cmd, bool connected, ulong ctx, ushort seq,
+        out (EncapsulationHeader, ReadOnlyMemory<byte>) reply)
+    {
+        reply = default;
+        var h = EncapsulationCodec.ReadHeader(frame);
+        if (h.Command != cmd) return false;
+
+        if (!connected)
+        {
+            if (h.SenderContext != ctx && h.SenderContext != 0) return false;
+            var p = EncapsulationCodec.ParseReply(frame, expectConnected: false);
+            reply = (p.Header, p.Cip);
+            return true;
+        }
+
+        // A connected error reply (status != 0) carries no readable sequence; with one request in flight
+        // and abandoned exchanges drained, it can only be ours — ParseReply throws BadReply for it.
+        var c = EncapsulationCodec.ParseReply(frame, expectConnected: true);
+        if (c.Sequence != seq) return false;
+        reply = (c.Header, c.Cip);
+        return true;
+    }
+
+    private async Task<byte[]> ReadFrameAsync(CancellationToken ct)
+    {
+        var head = new byte[EncapsulationHeader.Size];
+        await _stream.ReadExactlyAsync(head, ct).ConfigureAwait(false);
+        var len = head[2] | (head[3] << 8);
+        var frame = new byte[EncapsulationHeader.Size + len];
+        head.CopyTo(frame, 0);
+        await _stream.ReadExactlyAsync(frame.AsMemory(EncapsulationHeader.Size), ct).ConfigureAwait(false);
+        return frame;
+    }
+
+    private void ThrowIfUnusable()
+    {
+        if (Volatile.Read(ref _disposed) != 0)
+            throw new AbTransportException(AbStatus.BadConnection, "Session is disposed.");
+        if (_faulted)
+            throw new AbTransportException(AbStatus.BadConnection, "Session is faulted; reconnect.");
+    }
+
+    private void Fault()
+    {
+        _faulted = true;
+        try { _tcp.Close(); } catch { /* already closed */ }
+    }
+
+    /// <summary>Close without UnRegisterSession (registration never completed).</summary>
+    private void Abort()
+    {
+        Interlocked.Exchange(ref _disposed, 1);
+        Fault();
+    }
+
+    /// <summary>Best-effort UnRegisterSession, then close. Waits briefly for an in-flight exchange so the
+    /// UnRegisterSession frame is never interleaved with another frame; if the gate is not free in time
+    /// the socket is closed without it.</summary>
+    public async ValueTask DisposeAsync()
+    {
+        if (Interlocked.Exchange(ref _disposed, 1) != 0) return;
+        var acquired = await _gate.WaitAsync(UnregisterBudget).ConfigureAwait(false);
+        try
+        {
+            if (acquired && !_faulted && _tcp.Connected)
+            {
+                try
+                {
+                    var bytes = EncapsulationCodec.UnRegisterSession(SessionHandle, (ulong)Interlocked.Increment(ref _ctx));
+                    using var cts = new CancellationTokenSource(UnregisterBudget);
+                    await _stream.WriteAsync(bytes, cts.Token).ConfigureAwait(false);
+                    await _stream.FlushAsync(cts.Token).ConfigureAwait(false);
+                }
+                catch { /* best effort: the socket is closed next either way */ }
+            }
+        }
+        finally
+        {
+            Fault();
+            _tcp.Dispose();
+            if (acquired) _gate.Release();
+        }
+    }
+}
@@ -0,0 +1,111 @@
+using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Wire;
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip;
+
+/// <summary>Builds and parses EtherNet/IP encapsulation frames, including the Common Packet Format
+/// (CPF) items SendRRData / SendUnitData carry.</summary>
+public static class EncapsulationCodec
+{
+    /// <summary>CPF item: Null Address (unconnected messaging).</summary>
+    public const ushort ItemNullAddress = 0x0000;
+
+    /// <summary>CPF item: Connected Address (carries the O→T connection id).</summary>
+    public const ushort ItemConnectedAddress = 0x00A1;
+
+    /// <summary>CPF item: Connected Data (sequence count + CIP).</summary>
+    public const ushort ItemConnectedData = 0x00B1;
+
+    /// <summary>CPF item: Unconnected Data (CIP).</summary>
+    public const ushort ItemUnconnectedData = 0x00B2;
+
+    /// <summary>Largest CIP payload one frame can carry: the encapsulation length field (u16) also
+    /// covers the CPF overhead, so anything near 64 KiB cannot be framed.</summary>
+    public const int MaxCipPayload = ushort.MaxValue - 64;
+
+    private static CipWriter Header(ushort cmd, uint session, ulong ctx) =>
+        new CipWriter().U16(cmd).U16(0).U32(session).U32(0).U64(ctx).U32(0);
+
+    private static byte[] Finish(CipWriter w)
+    {
+        w.PatchU16(2, (ushort)(w.Length - EncapsulationHeader.Size));
+        return w.ToArray();
+    }
+
+    /// <summary>RegisterSession: protocol version 1, option flags 0.</summary>
+    public static byte[] RegisterSession(ulong ctx) =>
+        Finish(Header(EncapCommand.RegisterSession, 0, ctx).U16(1).U16(0));
+
+    /// <summary>UnRegisterSession: header only, no data.</summary>
+    public static byte[] UnRegisterSession(uint session, ulong ctx) =>
+        Finish(Header(EncapCommand.UnRegisterSession, session, ctx));
+
+    /// <summary>SendRRData (unconnected): Null Address item + Unconnected Data item carrying <paramref name="cip"/>.</summary>
+    public static byte[] SendRRData(uint session, ulong ctx, ReadOnlySpan<byte> cip, ushort timeoutSeconds) =>
+        Finish(Header(EncapCommand.SendRRData, session, ctx)
+            .U32(0).U16(timeoutSeconds).U16(2)
+            .U16(ItemNullAddress).U16(0)
+            .U16(ItemUnconnectedData).U16(CheckedLength(cip.Length)).Bytes(cip));
+
+    /// <summary>SendUnitData (connected): Connected Address item + Connected Data item carrying the
+    /// sequence count followed by <paramref name="cip"/>.</summary>
+    public static byte[] SendUnitData(uint session, ulong ctx, uint connectionId, ushort sequence, ReadOnlySpan<byte> cip) =>
+        Finish(Header(EncapCommand.SendUnitData, session, ctx)
+            .U32(0).U16(0).U16(2)
+            .U16(ItemConnectedAddress).U16(4).U32(connectionId)
+            .U16(ItemConnectedData).U16((ushort)(CheckedLength(cip.Length) + 2)).U16(sequence).Bytes(cip));
+
+    private static ushort CheckedLength(int length)
+    {
+        if (length > MaxCipPayload)
+            throw new AbTransportException(AbStatus.TooLarge, $"CIP payload of {length} bytes cannot be framed (max {MaxCipPayload}).");
+        return (ushort)length;
+    }
+
+    /// <summary>Read the 24-byte header at the start of <paramref name="b"/>.</summary>
+    public static EncapsulationHeader ReadHeader(ReadOnlySpan<byte> b)
+    {
+        var r = new CipReader(b);
+        return new EncapsulationHeader(r.U16(), r.U16(), r.U32(), r.U32(), r.U64(), r.U32());
+    }
+
+    /// <summary>Parse a reply frame. Returns the header plus the CIP payload of the data item
+    /// (0x00B2 unconnected / 0x00B1 connected, sequence count stripped) and, for a connected reply,
+    /// the echoed sequence count. RegisterSession / UnRegisterSession replies return an empty payload.
+    /// Throws <see cref="AbTransportException"/>(<see cref="AbStatus.BadReply"/>) on a non-zero
+    /// encapsulation status, a length mismatch, a truncated item list or a missing data item.</summary>
+    public static (EncapsulationHeader Header, ReadOnlyMemory<byte> Cip, ushort Sequence) ParseReply(
+        ReadOnlyMemory<byte> frame, bool expectConnected)
+    {
+        var h = ReadHeader(frame.Span);
+        if (h.Status != 0)
+            throw new AbTransportException(AbStatus.BadReply, $"Encapsulation status 0x{h.Status:X8} on command 0x{h.Command:X4}.");
+        if (h.Length != frame.Length - EncapsulationHeader.Size)
+            throw new AbTransportException(AbStatus.BadReply,
+                $"Encapsulation length mismatch: header says {h.Length}, frame carries {frame.Length - EncapsulationHeader.Size}.");
+        if (h.Command is EncapCommand.RegisterSession or EncapCommand.UnRegisterSession)
+            return (h, ReadOnlyMemory<byte>.Empty, 0);
+
+        var r = new CipReader(frame.Span[EncapsulationHeader.Size..]);
+        r.U32(); r.U16();                              // interface handle, timeout
+        var items = r.U16();
+        var offset = EncapsulationHeader.Size + 8;
+        var wanted = expectConnected ? ItemConnectedData : ItemUnconnectedData;
+        for (var i = 0; i < items; i++)
+        {
+            var type = r.U16();
+            var len = r.U16();
+            offset += 4;
+            r.Bytes(len);                              // bounds-checks the item against the frame
+            if (type == wanted)
+            {
+                if (!expectConnected) return (h, frame.Slice(offset, len), 0);
+                if (len < 2)
+                    throw new AbTransportException(AbStatus.BadReply, "Connected data item shorter than its sequence count.");
+                var seq = (ushort)(frame.Span[offset] | (frame.Span[offset + 1] << 8));
+                return (h, frame.Slice(offset + 2, len - 2), seq);
+            }
+            offset += len;
+        }
+        throw new AbTransportException(AbStatus.BadReply,
+            $"Reply to command 0x{h.Command:X4} carried no 0x{wanted:X4} data item.");
+    }
+}
@@ -0,0 +1,15 @@
+namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip;
+
+/// <summary>EtherNet/IP encapsulation command codes (CIP Vol. 2, Table 2-3.2).</summary>
+public static class EncapCommand
+{
+    public const ushort ListIdentity = 0x0063, RegisterSession = 0x0065, UnRegisterSession = 0x0066,
+        SendRRData = 0x006F, SendUnitData = 0x0070;
+}
+
+/// <summary>The fixed 24-byte EtherNet/IP encapsulation header (all fields little-endian).</summary>
+public readonly record struct EncapsulationHeader(
+    ushort Command, ushort Length, uint SessionHandle, uint Status, ulong SenderContext, uint Options)
+{
+    public const int Size = 24;
+}