Merge lane K: clients, CLI, AB transport (#747)

Claude-Session: https://claude.ai/code/session_01Ub9hGmYLQvn9XETzw7CMsT
This commit is contained in:
Joseph Doherty
2026-10-22 17:38:08 -04:00
53 changed files with 2339 additions and 510 deletions
@@ -44,17 +44,21 @@ public sealed class ProbeCommand : AbCipCommandBase
await using var driver = new AbCipDriver(options, DriverInstanceId);
try
{
await driver.InitializeAsync("{}", ct);
var snapshot = await driver.ReadAsync(["__probe"], ct);
var health = driver.GetHealth();
// Ctrl-C prints "Cancelled." instead of an OperationCanceledException stack trace (review 07 U7).
await RunCancellableAsync(console, ct, async () =>
{
await driver.InitializeAsync("{}", ct);
var snapshot = await driver.ReadAsync(["__probe"], ct);
var health = driver.GetHealth();
await console.Output.WriteLineAsync($"Gateway: {Gateway}");
await console.Output.WriteLineAsync($"Family: {Family}");
await console.Output.WriteLineAsync($"Health: {health.State}");
if (health.LastError is { } err)
await console.Output.WriteLineAsync($"Last error: {err}");
await console.Output.WriteLineAsync();
await console.Output.WriteLineAsync(SnapshotFormatter.Format(TagPath, snapshot[0]));
await console.Output.WriteLineAsync($"Gateway: {Gateway}");
await console.Output.WriteLineAsync($"Family: {Family}");
await console.Output.WriteLineAsync($"Health: {health.State}");
if (health.LastError is { } err)
await console.Output.WriteLineAsync($"Last error: {err}");
await console.Output.WriteLineAsync();
await console.Output.WriteLineAsync(SnapshotFormatter.Format(TagPath, snapshot[0]));
});
}
finally
{
@@ -45,9 +45,13 @@ public sealed class ReadCommand : AbCipCommandBase
await using var driver = new AbCipDriver(options, DriverInstanceId);
try
{
await driver.InitializeAsync("{}", ct);
var snapshot = await driver.ReadAsync([tagName], ct);
await console.Output.WriteLineAsync(SnapshotFormatter.Format(TagPath, snapshot[0]));
// Ctrl-C prints "Cancelled." instead of an OperationCanceledException stack trace (review 07 U7).
await RunCancellableAsync(console, ct, async () =>
{
await driver.InitializeAsync("{}", ct);
var snapshot = await driver.ReadAsync([tagName], ct);
await console.Output.WriteLineAsync(SnapshotFormatter.Format(TagPath, snapshot[0]));
});
}
finally
{
@@ -50,23 +50,22 @@ public sealed class SubscribeCommand : AbCipCommandBase
await using var driver = new AbCipDriver(options, DriverInstanceId);
ISubscriptionHandle? handle = null;
// The poll engine raises OnDataChange on a background thread: serialise its lines against each other and
// against this thread's own, and never let a failed console write fault the poll loop (review 07 U7).
var writeLock = new object();
try
{
await driver.InitializeAsync("{}", ct);
await console.Output.WriteLineAsync(
$"Subscribed to {TagPath} @ {IntervalMs}ms. Ctrl+C to stop.");
driver.OnDataChange += (_, e) =>
{
var line = $"[{DateTime.UtcNow:HH:mm:ss.fff}] " +
$"{e.FullReference} = {SnapshotFormatter.FormatValue(e.Snapshot.Value)} " +
$"({SnapshotFormatter.FormatStatus(e.Snapshot.StatusCode)})";
console.Output.WriteLine(line);
};
driver.OnDataChange += CreateChangeLineWriter(console.Output, writeLock);
handle = await driver.SubscribeAsync([tagName], TimeSpan.FromMilliseconds(IntervalMs), ct);
lock (writeLock)
{
console.Output.WriteLine($"Subscribed to {TagPath} @ {IntervalMs}ms. Ctrl+C to stop.");
}
try
{
await Task.Delay(System.Threading.Timeout.InfiniteTimeSpan, ct);
@@ -76,6 +75,10 @@ public sealed class SubscribeCommand : AbCipCommandBase
// Expected on Ctrl+C.
}
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
// Ctrl+C while still connecting/subscribing: a clean exit, not a stack trace (review 07 U7).
}
finally
{
if (handle is not null)
@@ -57,9 +57,13 @@ public sealed class WriteCommand : AbCipCommandBase
await using var driver = new AbCipDriver(options, DriverInstanceId);
try
{
await driver.InitializeAsync("{}", ct);
var results = await driver.WriteAsync([new WriteRequest(tagName, parsed)], ct);
await console.Output.WriteLineAsync(SnapshotFormatter.FormatWrite(TagPath, results[0]));
// Ctrl-C prints "Cancelled." instead of an OperationCanceledException stack trace (review 07 U7).
await RunCancellableAsync(console, ct, async () =>
{
await driver.InitializeAsync("{}", ct);
var results = await driver.WriteAsync([new WriteRequest(tagName, parsed)], ct);
await console.Output.WriteLineAsync(SnapshotFormatter.FormatWrite(TagPath, results[0]));
});
}
finally
{
@@ -32,27 +32,6 @@ public abstract class AbLegacyCommandBase : DriverCommandBase
"LogixPccc unverified).")]
public AbLegacyCliTransport Transport { get; init; } = AbLegacyCliTransport.Default;
/// <summary>
/// Runs a one-shot command's driver work so a Ctrl-C (cancellation of <paramref name="ct"/>) prints
/// <c>Cancelled.</c> instead of an unhandled <see cref="OperationCanceledException"/> stack trace (review 07 U7).
/// Any other exception — and an OCE the operator did not ask for — still propagates.
/// </summary>
/// <param name="console">The CLI console.</param>
/// <param name="ct">The console's Ctrl-C token.</param>
/// <param name="body">The command body.</param>
/// <returns>A task that completes when the body has finished or was cancelled.</returns>
internal static async ValueTask RunCancellableAsync(CliFx.Infrastructure.IConsole console, CancellationToken ct, Func<Task> body)
{
try
{
await body();
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
await console.Error.WriteLineAsync("Cancelled.");
}
}
/// <summary>Gets the per-operation timeout in milliseconds.</summary>
[CommandOption("timeout-ms", Description = "Per-operation timeout in ms (default 5000).")]
public int TimeoutMs { get; init; } = 5000;
@@ -79,5 +79,61 @@ public abstract class DriverCommandBase : ICommand
/// </summary>
protected static void FlushLogging() => Log.CloseAndFlush();
/// <summary>
/// Runs a one-shot command's driver work so a Ctrl-C (cancellation of <paramref name="ct"/>) prints
/// <c>Cancelled.</c> instead of an unhandled <see cref="OperationCanceledException"/> stack trace (review 07 U7).
/// Any other exception — and an OCE the operator did not ask for — still propagates.
/// </summary>
/// <param name="console">The CLI console.</param>
/// <param name="ct">The console's Ctrl-C token.</param>
/// <param name="body">The command body.</param>
/// <returns>A task that completes when the body has finished or was cancelled.</returns>
public static async ValueTask RunCancellableAsync(IConsole console, CancellationToken ct, Func<Task> body)
{
try
{
await body();
}
catch (OperationCanceledException) when (ct.IsCancellationRequested)
{
await console.Error.WriteLineAsync("Cancelled.");
}
}
/// <summary>
/// Builds the <c>OnDataChange</c> handler a <c>subscribe</c> command attaches: one
/// <c>[HH:mm:ss.fff] ref = value (status)</c> line per change. Writes are serialised on
/// <paramref name="writeLock"/> (the poll engine raises from a background thread, concurrently with the
/// command thread's own lines), and a failing console write (closed pipe, redirected stream error) is logged
/// and swallowed so it cannot fault the driver's poll loop (review 07 U7).
/// </summary>
/// <param name="output">The console output writer.</param>
/// <param name="writeLock">The lock every console write of the command takes.</param>
/// <returns>The event handler.</returns>
public static EventHandler<ZB.MOM.WW.OtOpcUa.Core.Abstractions.DataChangeEventArgs> CreateChangeLineWriter(
TextWriter output, object writeLock)
{
ArgumentNullException.ThrowIfNull(output);
ArgumentNullException.ThrowIfNull(writeLock);
return (_, e) =>
{
try
{
var line = $"[{DateTime.UtcNow:HH:mm:ss.fff}] " +
$"{e.FullReference} = {SnapshotFormatter.FormatValue(e.Snapshot.Value)} " +
$"({SnapshotFormatter.FormatStatus(e.Snapshot.StatusCode)})";
lock (writeLock)
{
output.WriteLine(line);
}
}
catch (Exception ex)
{
Log.Logger.Warning(ex, "subscribe: console write failed for {Tag}; continuing the poll loop.",
e.FullReference);
}
};
}
private bool _loggingConfigured;
}
@@ -101,7 +101,11 @@ public static class SnapshotFormatter
return sb.ToString().TrimEnd();
}
/// <summary>Formats a value for console output, handling null, bool, string, and formattable types.</summary>
/// <summary>
/// Formats a value for console output, handling null, bool, string, formattable and array/list types. An
/// array renders its elements (<c>[1, 2, 3]</c>, each formatted as a scalar) rather than its CLR type name
/// (<c>System.Int32[]</c>).
/// </summary>
/// <param name="value">The value to format.</param>
/// <returns>A formatted string representation of the value.</returns>
public static string FormatValue(object? value) => value switch
@@ -110,9 +114,23 @@ public static class SnapshotFormatter
bool b => b ? "true" : "false",
string s => $"\"{s}\"",
IFormattable f => f.ToString(null, CultureInfo.InvariantCulture),
System.Collections.IEnumerable items => FormatSequence(items),
_ => value.ToString() ?? "<null>",
};
private static string FormatSequence(System.Collections.IEnumerable items)
{
var sb = new System.Text.StringBuilder("[");
var first = true;
foreach (var item in items)
{
if (!first) sb.Append(", ");
sb.Append(FormatValue(item));
first = false;
}
return sb.Append(']').ToString();
}
/// <summary>Formats an OPC UA status code as a hexadecimal string with named severity classification.</summary>
/// <param name="statusCode">The OPC UA status code to format.</param>
/// <returns>A formatted status string like "0x00000000 (Good)" or "0x80050000 (BadCommunicationError)".</returns>
@@ -0,0 +1,12 @@
namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport;
/// <summary>
/// An <see cref="AbTransportException"/> whose status came OFF THE WIRE: the device (or the bridge's Execute PCCC
/// object) replied with a non-zero CIP envelope status or PCCC STS. It lets a driver tell a refusal the device
/// ANSWERED from the same <see cref="AbStatus"/> raised locally before anything was sent — most importantly
/// <see cref="AbStatus.Unsupported"/>, which a native runtime also throws for operations it refuses pre-wire
/// (review lane K I2).
/// </summary>
/// <param name="status">The status mapped from the reply.</param>
/// <param name="message">The exception message.</param>
public sealed class AbDeviceReplyException(AbStatus status, string message) : AbTransportException(status, message);
@@ -1,39 +1,81 @@
namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport;
/// <summary>
/// Maps wire statuses (CIP general/extended, PCCC STS/EXT STS) to the transport-neutral <see cref="AbStatus"/> the
/// drivers classify. The class an <see cref="AbStatus"/> lands in is load-bearing for driver health (#527/#737/#738):
/// an ANSWERED refusal (NotFound / NotAllowed / OutOfBounds / TooLarge / TooSmall / TypeMismatch /
/// DeviceStateConflict, and for AbCip Unsupported) proves the device replied; <see cref="AbStatus.BadConnection"/>
/// is the no-reply comms class; <see cref="AbStatus.BadReply"/> / <see cref="AbStatus.Partial"/> are a reply that
/// cannot be classified. Every code not named below is deliberately <see cref="AbStatus.BadReply"/>.
/// </summary>
public static class AbStatusCodes
{
/// <summary>Maps a CIP general status (and optional extended status words) to an <see cref="AbStatus"/>.</summary>
/// <param name="general">The CIP general status byte.</param>
/// <param name="extended">The extended status words, if any.</param>
/// <returns>The mapped status.</returns>
/// <remarks>
/// 0x01 (connection failure, incl. a routing hop's unconnected-send timeout) and 0x07 (connection lost) stay
/// <see cref="AbStatus.BadReply"/> on purpose: a bridge reports a dead END device that way, so they are neither an
/// answer nor proof of no reply (AbLegacy #738 "opaque"). A Forward Open refused with one of the request-level
/// codes (0x1C / 0x20 / 0xFF-0x2104) stays a connection failure — see
/// <c>ConnectionManagerServices.ParseForwardOpenReply</c>.
/// </remarks>
public static AbStatus FromCip(byte general, ReadOnlySpan<ushort> extended) => general switch
{
0x00 or 0x0B => AbStatus.Ok,
0x04 or 0x05 or 0x16 => AbStatus.NotFound,
0x06 => AbStatus.Partial,
0x08 => AbStatus.Unsupported,
0x0A => AbStatus.OutOfBounds,
0x0C or 0x10 => AbStatus.DeviceStateConflict,
0x0E or 0x0F => AbStatus.NotAllowed,
0x13 => AbStatus.TooSmall,
0x15 => AbStatus.TooLarge,
0xFF when extended.Length > 0 && extended[0] == 0x2105 => AbStatus.OutOfBounds,
0xFF when extended.Length > 0 && extended[0] == 0x2107 => AbStatus.TypeMismatch,
0x00 or 0x0B => AbStatus.Ok, // success / already in requested mode
0x04 or 0x05 or 0x16 => AbStatus.NotFound, // path segment error / path unknown / object does not exist
0x06 => AbStatus.Partial, // partial transfer (fragmented read continues)
0x08 => AbStatus.Unsupported, // service not supported
0x0A => AbStatus.OutOfBounds, // attribute list error
0x0C or 0x10 => AbStatus.DeviceStateConflict, // object state conflict / device state conflict
0x0E or 0x0F => AbStatus.NotAllowed, // attribute not settable / privilege violation
0x13 => AbStatus.TooSmall, // not enough data
0x15 => AbStatus.TooLarge, // too much data
// Lane K F1 (#737): the controller answered and refused this one request — not a garbled reply, so not a
// comms fault that degrades the driver and evicts the runtime.
0x1C => AbStatus.Unsupported, // missing attribute list entry data
0x20 => AbStatus.Unsupported, // invalid parameter
0xFF when extended.Length > 0 && extended[0] == 0x2104 => AbStatus.OutOfBounds, // offset beyond end of tag
0xFF when extended.Length > 0 && extended[0] == 0x2105 => AbStatus.OutOfBounds, // beyond end of the object
0xFF when extended.Length > 0 && extended[0] == 0x2107 => AbStatus.TypeMismatch, // data type mismatch
_ => AbStatus.BadReply,
};
/// <summary>
/// True for the CIP statuses Lane K F1 moved from <see cref="AbStatus.BadReply"/> to a request-level refusal.
/// A Forward Open refused with one of them is still a connection failure.
/// </summary>
/// <param name="general">The CIP general status byte.</param>
/// <param name="extended">The extended status words, if any.</param>
/// <returns><see langword="true"/> for 0x1C, 0x20 and 0xFF/0x2104.</returns>
internal static bool IsRequestLevelRefusal(byte general, ReadOnlySpan<ushort> extended) =>
general is 0x1C or 0x20 || (general == 0xFF && extended.Length > 0 && extended[0] == 0x2104);
/// <summary>Maps a PCCC status byte (and optional extended status) to an <see cref="AbStatus"/>.</summary>
/// <param name="sts">The PCCC STS byte.</param>
/// <param name="extSts">The extended status byte, when present.</param>
/// <returns>The mapped status.</returns>
/// <remarks>
/// Per the DF1/PCCC STS table (1770-6.5.16). Lane K F3 (AbLegacy F8, review 07 r0 I4): 0x20, 0x30 and 0x90 are
/// reported by the HOST or a BRIDGE (a DHRIO, a 1761-NET-ENI) when the END device cannot be reached, so they are
/// the no-reply comms class. Mapping them to an answered refusal made a dead DH+/DH-485 node behind a bridge count
/// as reachable forever — no fail-fast backoff, no comms-loss watchdog.
/// </remarks>
public static AbStatus FromPccc(byte sts, byte? extSts) => sts switch
{
0x00 => AbStatus.Ok,
0x10 => AbStatus.Unsupported,
0x20 => AbStatus.NotFound,
0x30 => AbStatus.NotAllowed,
0x40 or 0x50 => AbStatus.DeviceStateConflict,
0xF0 when extSts is 0x06 or 0x07 => AbStatus.NotFound,
_ => AbStatus.BadReply,
0x10 => AbStatus.Unsupported, // illegal command or format
0x20 => AbStatus.BadConnection, // host has a problem and will not communicate
0x30 => AbStatus.BadConnection, // remote node host missing, disconnected or shut down
0x40 => AbStatus.DeviceStateConflict, // host could not complete: hardware fault
0x50 => AbStatus.NotFound, // addressing problem or memory protect rungs
0x60 => AbStatus.NotAllowed, // function not allowed due to command protection
0x70 => AbStatus.DeviceStateConflict, // processor is in Program mode
0x90 => AbStatus.BadConnection, // remote node cannot buffer command
0xB0 => AbStatus.DeviceStateConflict, // remote node problem due to download
0xF0 when extSts is 0x06 => AbStatus.NotFound, // EXT: address does not point to something usable
0xF0 when extSts is 0x07 => AbStatus.OutOfBounds, // EXT: file is wrong size
_ => AbStatus.BadReply, // 0x80, 0xA0, 0xC0, other EXT STS: unclassified
};
}
@@ -85,8 +85,15 @@ public static class ConnectionManagerServices
public static ForwardOpenReply ParseForwardOpenReply(CipReply reply)
{
if (reply.GeneralStatus != 0)
throw new ForwardOpenRefusedException(reply.Status == AbStatus.BadReply ? AbStatus.BadConnection : reply.Status,
reply.GeneralStatus, reply.ExtendedStatus);
{
// An unclassified refusal, or one of the request-level codes AbStatusCodes maps to a per-request refusal
// for TAG services (0x1C / 0x20 / 0xFF-0x2104), is a connection failure here: no tag can be served.
var status = reply.Status == AbStatus.BadReply
|| AbStatusCodes.IsRequestLevelRefusal(reply.GeneralStatus, reply.ExtendedStatus)
? AbStatus.BadConnection
: reply.Status;
throw new ForwardOpenRefusedException(status, reply.GeneralStatus, reply.ExtendedStatus);
}
var r = new CipReader(reply.Data.Span);
var ot = r.U32(); var to = r.U32(); r.U16(); r.U16(); r.U32();
return new ForwardOpenReply(ot, to, r.U32(), r.U32());
@@ -47,6 +47,10 @@ public sealed class EncapsulationClient : IAsyncDisposable
private readonly SemaphoreSlim _gate = new(1, 1);
private long _ctx;
private ushort _seq; // connected sequence count; only touched under _gate
// R07:L5: the 24-byte header of the frame being read. Frames are read only by RunExchangeAsync, which holds _gate
// for the whole exchange, so one buffer per session serves every read (it used to be a fresh array per reply).
private readonly byte[] _head = new byte[EncapsulationHeader.Size];
private volatile bool _faulted;
private int _disposed;
@@ -282,11 +286,12 @@ public sealed class EncapsulationClient : IAsyncDisposable
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);
// Only called under _gate (RunExchangeAsync), so the shared header buffer is never read concurrently. The frame
// itself is allocated exactly: the matched reply's CIP payload is returned as a slice of it and outlives the call.
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);
_head.CopyTo(frame, 0);
await _stream.ReadExactlyAsync(frame.AsMemory(EncapsulationHeader.Size), ct).ConfigureAwait(false);
return frame;
}
@@ -1,3 +1,4 @@
using System.Buffers;
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Logix;
@@ -52,34 +53,61 @@ public sealed class LogixTagClient(RequestScheduler scheduler, ICipChannel chann
var reply = await _scheduler.SendAsync(LogixServices.ReadTagFragmented(path, elements, 0), ct,
expectedReplyBytes: ExpectedReadReplyBytes(elements, elementBytes, 0)).ConfigureAwait(false);
var (type, handle, first) = LogixServices.ParseReadReply(reply); // throws on any status but 0x00/0x06
var data = new List<byte>(first.Length);
data.AddRange(first.Span);
var lastChunk = first.Length;
while (reply.GeneralStatus == StatusPartial)
// R07:L5: the common single-reply read is one copy of the reply data (it used to go through a List<byte> and a
// second ToArray); a fragmented read reassembles in a pooled buffer and copies out once at the end.
if (reply.GeneralStatus != StatusPartial) return (type, handle, first.ToArray());
var data = ArrayPool<byte>.Shared.Rent(Math.Max(first.Length * 4, 1024));
var count = 0;
try
{
if (lastChunk == 0)
throw new AbTransportException(AbStatus.BadReply, "Partial read reply carried no data; the fragment loop cannot progress.");
reply = await _scheduler.SendAsync(
LogixServices.ReadTagFragmented(path, elements, (uint)data.Count), ct,
expectedReplyBytes: ExpectedReadReplyBytes(elements, elementBytes, data.Count)).ConfigureAwait(false);
var (fragmentType, fragmentHandle, chunk) = LogixServices.ParseReadReply(reply);
if (fragmentType != type)
throw new AbTransportException(AbStatus.BadReply,
$"Read fragment type 0x{fragmentType:X4} differs from the first reply's 0x{type:X4}.");
// A mid-read change of the structure handle means the tag's type changed under us (a
// download / online edit); stitching the halves would yield a value of neither type.
if (fragmentHandle != handle)
throw new AbTransportException(AbStatus.BadReply,
$"Read fragment structure handle {FormatHandle(fragmentHandle)} differs from the first reply's {FormatHandle(handle)}; the tag changed mid-read.");
data.AddRange(chunk.Span);
lastChunk = chunk.Length;
if (data.Count > MaxReadBytes)
throw new AbTransportException(AbStatus.TooLarge,
$"Fragmented read exceeded {MaxReadBytes} bytes; the controller kept answering Partial.");
}
Append(ref data, ref count, first.Span);
var lastChunk = first.Length;
return (type, handle, data.ToArray());
while (reply.GeneralStatus == StatusPartial)
{
if (lastChunk == 0)
throw new AbTransportException(AbStatus.BadReply, "Partial read reply carried no data; the fragment loop cannot progress.");
reply = await _scheduler.SendAsync(
LogixServices.ReadTagFragmented(path, elements, (uint)count), ct,
expectedReplyBytes: ExpectedReadReplyBytes(elements, elementBytes, count)).ConfigureAwait(false);
var (fragmentType, fragmentHandle, chunk) = LogixServices.ParseReadReply(reply);
if (fragmentType != type)
throw new AbTransportException(AbStatus.BadReply,
$"Read fragment type 0x{fragmentType:X4} differs from the first reply's 0x{type:X4}.");
// A mid-read change of the structure handle means the tag's type changed under us (a
// download / online edit); stitching the halves would yield a value of neither type.
if (fragmentHandle != handle)
throw new AbTransportException(AbStatus.BadReply,
$"Read fragment structure handle {FormatHandle(fragmentHandle)} differs from the first reply's {FormatHandle(handle)}; the tag changed mid-read.");
Append(ref data, ref count, chunk.Span);
lastChunk = chunk.Length;
if (count > MaxReadBytes)
throw new AbTransportException(AbStatus.TooLarge,
$"Fragmented read exceeded {MaxReadBytes} bytes; the controller kept answering Partial.");
}
return (type, handle, data.AsSpan(0, count).ToArray());
}
finally
{
ArrayPool<byte>.Shared.Return(data);
}
}
/// <summary>Appends <paramref name="chunk"/> to a pooled buffer, growing it (rent bigger, copy, return) as needed.</summary>
private static void Append(ref byte[] buffer, ref int count, ReadOnlySpan<byte> chunk)
{
if (count + chunk.Length > buffer.Length)
{
var bigger = ArrayPool<byte>.Shared.Rent(Math.Max(buffer.Length * 2, count + chunk.Length));
buffer.AsSpan(0, count).CopyTo(bigger);
ArrayPool<byte>.Shared.Return(buffer);
buffer = bigger;
}
chunk.CopyTo(buffer.AsSpan(count));
count += chunk.Length;
}
private static string FormatHandle(ushort? handle) => handle is { } h ? $"0x{h:X4}" : "(none)";
@@ -113,7 +113,7 @@ public static class PcccServices
public static ReadOnlyMemory<byte> ParseReply(CipReply reply, ushort expectedTns, byte[]? expectedRequestorId = null)
{
if (reply.GeneralStatus != 0)
throw new AbTransportException(reply.Status, $"Execute PCCC CIP status 0x{reply.GeneralStatus:X2}.");
throw new AbDeviceReplyException(reply.Status, $"Execute PCCC CIP status 0x{reply.GeneralStatus:X2}.");
var r = new CipReader(reply.Data.Span);
var idLen = r.U8();
if (idLen < 1) throw new AbTransportException(AbStatus.BadReply, "Empty requestor id in PCCC reply.");
@@ -131,7 +131,7 @@ public static class PcccServices
throw new AbTransportException(AbStatus.BadReply, $"PCCC TNS {tns} != {expectedTns}.");
byte? ext = sts == 0xF0 ? r.U8() : null;
if (sts != 0)
throw new AbTransportException(AbStatusCodes.FromPccc(sts, ext),
throw new AbDeviceReplyException(AbStatusCodes.FromPccc(sts, ext),
$"PCCC STS 0x{sts:X2}{(ext is { } e ? $" EXT 0x{e:X2}" : "")}.");
return reply.Data[r.Position..];
}
@@ -1,17 +1,38 @@
using System.Buffers;
using System.Buffers.Binary;
namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Wire;
/// <summary>Growable little-endian writer for CIP / EIP frames.</summary>
/// <remarks>
/// R07:L5: the working buffer is rented from <see cref="ArrayPool{T}.Shared"/> (a 256-byte array per request used
/// to be allocated and thrown away) and handed back by <see cref="ToArray"/>, which therefore COMPLETES the writer:
/// the result is the caller's own array, a second <see cref="ToArray"/> returns another copy of it, and any further
/// write throws <see cref="InvalidOperationException"/> (the buffer belongs to the pool again). A writer abandoned
/// before <see cref="ToArray"/> (a builder that threw) simply leaves its buffer to the GC.
/// </remarks>
public sealed class CipWriter
{
private byte[] _buf = new byte[256];
private const int InitialCapacity = 256;
private byte[]? _buf = ArrayPool<byte>.Shared.Rent(InitialCapacity);
private byte[]? _result;
/// <summary>Gets the number of bytes written so far.</summary>
public int Length { get; private set; }
private byte[] Buffer => _buf ?? throw new InvalidOperationException(
"This CipWriter was completed by ToArray(); its buffer has been returned to the pool.");
private Span<byte> Take(int n)
{
if (Length + n > _buf.Length) Array.Resize(ref _buf, Math.Max(_buf.Length * 2, Length + n));
var s = _buf.AsSpan(Length, n);
var buf = Buffer;
if (Length + n > buf.Length)
{
var bigger = ArrayPool<byte>.Shared.Rent(Math.Max(buf.Length * 2, Length + n));
buf.AsSpan(0, Length).CopyTo(bigger);
ArrayPool<byte>.Shared.Return(buf);
_buf = buf = bigger;
}
var s = buf.AsSpan(Length, n);
Length += n;
return s;
}
@@ -45,14 +66,27 @@ public sealed class CipWriter
/// <param name="v">The replacement value.</param>
/// <exception cref="ArgumentOutOfRangeException">The two bytes at <paramref name="offset"/> are not already
/// written (an offset in the spare buffer capacity would otherwise be patched silently and lost).</exception>
/// <exception cref="InvalidOperationException">The writer was completed by <see cref="ToArray"/>.</exception>
public void PatchU16(int offset, ushort v)
{
var buf = Buffer;
if (offset < 0 || offset > Length - 2)
throw new ArgumentOutOfRangeException(nameof(offset), offset, $"Cannot patch 2 bytes at {offset}: only {Length} bytes written.");
BinaryPrimitives.WriteUInt16LittleEndian(_buf.AsSpan(offset, 2), v);
BinaryPrimitives.WriteUInt16LittleEndian(buf.AsSpan(offset, 2), v);
}
/// <summary>Copies the written bytes to a new array.</summary>
/// <returns>The bytes written so far.</returns>
public byte[] ToArray() => _buf.AsSpan(0, Length).ToArray();
/// <summary>
/// Copies the written bytes to a new array and returns the working buffer to the pool, completing the writer
/// (see the remarks on <see cref="CipWriter"/>).
/// </summary>
/// <returns>The bytes written, in an array the caller owns.</returns>
public byte[] ToArray()
{
if (_result is { } done) return (byte[])done.Clone();
var buf = Buffer;
_result = buf.AsSpan(0, Length).ToArray();
_buf = null;
ArrayPool<byte>.Shared.Return(buf);
return _result;
}
}
@@ -13,10 +13,10 @@ namespace ZB.MOM.WW.OtOpcUa.Driver.AbCip;
/// never writes <c>Degraded</c>, never opens the device-wide connect backoff, and warns once per tag. Everything
/// else (timeout, bad connection, a reply that failed to parse, partial, abort, …) is a comms fault.
/// <para>
/// The native transport currently folds CIP general statuses 0x1C, 0x20 and 0xFF/0x2104 into
/// <see cref="AbStatus.BadReply"/> (<c>AbStatusCodes.FromCip</c>, owned by <c>Driver.Ab.Transport</c>), which
/// this classification cannot tell from a garbled reply; until the transport maps them explicitly they stay
/// comms faults.
/// The native transport maps CIP general statuses 0x1C (missing attribute list entry) and 0x20 (invalid
/// parameter) to <see cref="AbStatus.Unsupported"/> and 0xFF/0x2104 (offset beyond end) to
/// <see cref="AbStatus.OutOfBounds"/> (<c>AbStatusCodes.FromCip</c>, Lane K F1), so they classify here as
/// device refusals. A Forward Open refused with one of them stays a connection failure.
/// </para>
/// </remarks>
internal static class AbCipRefusals
@@ -789,9 +789,9 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.BadNotSupported, null, now);
return (OpVerdict.Neutral, null);
}
catch (AbTransportException ex) when (IsDeviceRefusal(ex.Status))
catch (AbTransportException ex) when (IsAnsweredRefusal(ex))
{
// The native runtime raises a device's refusal (e.g. STS 0x20 "file not found") as an exception.
// The native runtime raises a device's refusal (e.g. STS 0x50 "addressing problem") as an exception.
WarnRefusalOnce(device, def, "read", ex.Message, AbLegacyStatusMapper.MapStatus(ex.Status));
results[item.Index] = new DataValueSnapshot(null, AbLegacyStatusMapper.MapStatus(ex.Status), null, now);
return (OpVerdict.Answered, null);
@@ -838,10 +838,10 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
/// reachability (backoff reset, <c>LastSuccessfulRead</c>, Healthy).
/// </summary>
/// <remarks>
/// ⚠️ Inherits the native transport's PCCC STS table (<c>AbStatusCodes.FromPccc</c>, Lane K): it maps STS 0x30
/// ("remote node missing" — a dead DH+/DH-485 node behind a bridge) and 0x20 to <c>NotAllowed</c>/<c>NotFound</c>,
/// so on the NATIVE transport such a dead node still reads as answered. Follow-up F8 (review 07 r0 I4); this
/// classification needs no change once it lands.
/// Inherits the native transport's PCCC STS table (<c>AbStatusCodes.FromPccc</c>). Since Lane K F3 (F8, review
/// 07 r0 I4) STS 0x20 / 0x30 / 0x90 — the host or a bridge reporting an unreachable END device — map to
/// <see cref="AbStatus.BadConnection"/> (no reply), so a dead DH+/DH-485 node behind a bridge is a comms fault,
/// not an answer (<c>NativePcccBridgeStatusHealthTests</c>).
/// </remarks>
/// <param name="status">The transport status.</param>
/// <returns><see langword="true"/> for a device refusal.</returns>
@@ -1035,7 +1035,7 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
results[i] = new WriteResult(AbLegacyStatusMapper.BadCommunicationError);
outcome.FaultMessage ??= tex.Message;
}
catch (AbTransportException ex) when (IsDeviceRefusal(ex.Status))
catch (AbTransportException ex) when (IsAnsweredRefusal(ex))
{
device.RecordAnswered();
outcome.AnyAnswered = true;
@@ -1210,7 +1210,7 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
try { probeRuntime?.Dispose(); } catch { }
probeRuntime = null;
state.ProbeInitialized = false;
if (ex is AbTransportException ate && IsDeviceRefusal(ate.Status)) state.RecordAnswered();
if (IsAnsweredRefusal(ex)) state.RecordAnswered();
else if (ex is not AbTransportException { Status: var st } || IsNoReply(st)) state.RecordCommsFault(Clock());
}
@@ -1478,7 +1478,7 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
{
return (null, AbLegacyStatusMapper.BadNotSupported, false, false);
}
catch (AbTransportException ex) when (IsDeviceRefusal(ex.Status))
catch (AbTransportException ex) when (IsAnsweredRefusal(ex))
{
// The device answered: the index file/element does not exist. Per-tag, not a comms fault.
return (null, AbLegacyStatusMapper.MapStatus(ex.Status), false, false);
@@ -1617,10 +1617,26 @@ public sealed partial class AbLegacyDriver : IDriver, IReadable, IWritable, ITag
/// all PLC-5 writes), and libplctag's ErrorUnsupported maps to it too. It says nothing about the
/// device's reachability, so it must not evict the handle or degrade driver health.
/// </summary>
/// <remarks>
/// An Unsupported that came OFF THE WIRE (<see cref="AbDeviceReplyException"/>: CIP 0x08 / 0x1C / 0x20 on the
/// Execute PCCC envelope, PCCC STS 0x10) is NOT local — the device answered — and is classified by
/// <see cref="IsAnsweredRefusal"/> instead (review lane K I2).
/// </remarks>
/// <param name="ex">The exception to classify.</param>
/// <returns><see langword="true"/> if the exception is a local refusal that never reached the wire.</returns>
internal static bool IsLocalRefusal(Exception ex) =>
ex is AbTransportException { Status: AbStatus.Unsupported };
ex is AbTransportException { Status: AbStatus.Unsupported } and not AbDeviceReplyException;
/// <summary>
/// True for an exception proving the device ANSWERED and refused this operation: a device-refusal status
/// (<see cref="IsDeviceRefusal"/>), or an <see cref="AbStatus.Unsupported"/> that came off the wire
/// (<see cref="AbDeviceReplyException"/>). Per-tag Bad, keeps the handle, advances <c>LastSuccessfulRead</c>.
/// </summary>
/// <param name="ex">The exception to classify.</param>
/// <returns><see langword="true"/> for an answered refusal.</returns>
internal static bool IsAnsweredRefusal(Exception ex) =>
ex is AbTransportException ate
&& (IsDeviceRefusal(ate.Status) || ate is AbDeviceReplyException { Status: AbStatus.Unsupported });
/// <summary>
/// Initializes a freshly created runtime. A NATIVE runtime whose init (one read) is refused locally is
@@ -1,3 +1,4 @@
using System.Buffers;
namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus;
/// <summary>
@@ -55,12 +56,33 @@ public static class ModbusRtuFraming
/// <returns>A new array: <c>[unitId, ...pdu, crcLo, crcHi]</c>.</returns>
public static byte[] BuildAdu(byte unitId, ReadOnlySpan<byte> pdu)
{
var frame = new byte[1 + pdu.Length];
frame[0] = unitId;
pdu.CopyTo(frame.AsSpan(1));
return ModbusCrc.Append(frame);
var frame = new byte[AduLength(pdu.Length)];
WriteAdu(unitId, pdu, frame);
return frame;
}
/// <summary>The length of the RTU ADU for a PDU of <paramref name="pduLength"/> bytes (unit id + PDU + CRC).</summary>
/// <param name="pduLength">The bare PDU length.</param>
/// <returns>The ADU length.</returns>
internal static int AduLength(int pduLength) => 1 + pduLength + 2;
/// <summary>Writes the RTU ADU into <paramref name="destination"/> (exactly <see cref="AduLength"/> bytes).</summary>
/// <param name="unitId">The RTU slave/unit address.</param>
/// <param name="pdu">The bare PDU.</param>
/// <param name="destination">The ADU buffer, exactly <see cref="AduLength"/>(<paramref name="pdu"/>.Length) long.</param>
internal static void WriteAdu(byte unitId, ReadOnlySpan<byte> pdu, Span<byte> destination)
{
destination[0] = unitId;
pdu.CopyTo(destination[1..]);
var body = destination[..(1 + pdu.Length)];
var crc = ModbusCrc.Compute(body);
destination[1 + pdu.Length] = (byte)(crc & 0xFF); // CRC low byte first
destination[2 + pdu.Length] = (byte)(crc >> 8);
}
/// <summary>The largest RTU response frame: addr + fc + byteCount + 255 data bytes + CRC.</summary>
private const int MaxResponseFrame = 3 + 255 + 2;
/// <summary>
/// Reads a single RTU response frame from <paramref name="stream"/> and returns its bare PDU
/// (<c>[fc, ...data]</c>), with the leading unit-id and trailing CRC stripped.
@@ -82,58 +104,56 @@ public static class ModbusRtuFraming
public static async Task<byte[]> ReadResponsePduAsync(
Stream stream, byte expectedUnit, CancellationToken ct)
{
// Header: addr(1) + fc(1). The function code selects how many more bytes to read.
var header = new byte[2];
await ReadExactlyAsync(stream, header, ct).ConfigureAwait(false);
var fc = header[1];
// R07:L5: the whole frame is read into ONE pooled buffer and validated in place; only the returned PDU is
// allocated (it used to be five arrays: header, byteCount, rest, the reassembled frame and the PDU).
var frame = ArrayPool<byte>.Shared.Rent(MaxResponseFrame);
try
{
// Header: addr(1) + fc(1). The function code selects how many more bytes to read.
await ReadExactlyAsync(stream, frame.AsMemory(0, 2), ct).ConfigureAwait(false);
var fc = frame[1];
int trailing; // bytes after addr+fc, up to and including the 2 CRC bytes.
if ((fc & 0x80) != 0)
{
// Exception frame: excCode(1) + CRC(2).
trailing = 1 + 2;
}
else if (fc is 0x01 or 0x02 or 0x03 or 0x04 or 0x17)
{
// Read response (and FC23 Read/Write Multiple, whose reply is a read): byteCount(1) then
// byteCount data bytes + CRC(2).
var bc = new byte[1];
await ReadExactlyAsync(stream, bc, ct).ConfigureAwait(false);
var byteCount = bc[0];
var rest = new byte[byteCount + 2];
await ReadExactlyAsync(stream, rest, ct).ConfigureAwait(false);
return ValidateAndStrip(expectedUnit, header, bc, rest);
}
else
{
// Fixed 4-byte echo: correct for the write FCs this driver emits (05/06/0F/10). FC23 is
// variable-length and handled above; any NEW variable-length response FC must be added there
// or it is mis-sized here and surfaces as a desync.
trailing = 4 + 2;
}
int length; // the whole frame, addr through CRC
var already = 2; // bytes of the frame read so far
if ((fc & 0x80) != 0)
{
// Exception frame: excCode(1) + CRC(2).
length = 2 + 1 + 2;
}
else if (fc is 0x01 or 0x02 or 0x03 or 0x04 or 0x17)
{
// Read response (and FC23 Read/Write Multiple, whose reply is a read): byteCount(1) then
// byteCount data bytes + CRC(2).
await ReadExactlyAsync(stream, frame.AsMemory(2, 1), ct).ConfigureAwait(false);
length = 3 + frame[2] + 2;
already = 3;
}
else
{
// Fixed 4-byte echo: correct for the write FCs this driver emits (05/06/0F/10). FC23 is
// variable-length and handled above; any NEW variable-length response FC must be added there
// or it is mis-sized here and surfaces as a desync.
length = 2 + 4 + 2;
}
var tail = new byte[trailing];
await ReadExactlyAsync(stream, tail, ct).ConfigureAwait(false);
return ValidateAndStrip(expectedUnit, header, ReadOnlySpan<byte>.Empty, tail);
await ReadExactlyAsync(stream, frame.AsMemory(already, length - already), ct).ConfigureAwait(false);
return ValidateAndStrip(expectedUnit, frame.AsSpan(0, length));
}
finally
{
ArrayPool<byte>.Shared.Return(frame);
}
}
/// <summary>
/// Assembles the full frame from its pieces, validates the trailing CRC over everything
/// except the CRC bytes, then validates the response address against
/// <paramref name="expectedUnit"/> and strips <c>addr</c> + <c>CRC</c> to return the bare PDU.
/// Validates the trailing CRC over everything except the CRC bytes, then validates the response address
/// against <paramref name="expectedUnit"/> and strips <c>addr</c> + <c>CRC</c> to return the bare PDU.
/// A CRC-valid exception PDU throws <see cref="ModbusException"/>.
/// </summary>
private static byte[] ValidateAndStrip(
byte expectedUnit, ReadOnlySpan<byte> header, ReadOnlySpan<byte> middle, ReadOnlySpan<byte> tail)
private static byte[] ValidateAndStrip(byte expectedUnit, ReadOnlySpan<byte> frame)
{
// Reassemble the whole frame: header(addr,fc) + middle(optional byteCount) + tail.
var frame = new byte[header.Length + middle.Length + tail.Length];
header.CopyTo(frame);
middle.CopyTo(frame.AsSpan(header.Length));
tail.CopyTo(frame.AsSpan(header.Length + middle.Length));
// CRC covers everything except the trailing 2 CRC bytes.
var body = frame.AsSpan(0, frame.Length - 2);
var body = frame[..^2];
var expected = ModbusCrc.Compute(body);
var actual = (ushort)(frame[^2] | (frame[^1] << 8));
if (actual != expected)
@@ -172,12 +192,12 @@ public static class ModbusRtuFraming
/// <see cref="ModbusTransportDesyncException"/> so a length-less RTU frame that arrives
/// short is classified as a desync rather than a bare <see cref="EndOfStreamException"/>.
/// </summary>
private static async Task ReadExactlyAsync(Stream stream, byte[] buf, CancellationToken ct)
private static async Task ReadExactlyAsync(Stream stream, Memory<byte> buf, CancellationToken ct)
{
var read = 0;
while (read < buf.Length)
{
var n = await stream.ReadAsync(buf.AsMemory(read), ct).ConfigureAwait(false);
var n = await stream.ReadAsync(buf[read..], ct).ConfigureAwait(false);
if (n == 0)
throw new ModbusTransportDesyncException(
ModbusTransportDesyncException.DesyncReason.TruncatedFrame,
@@ -1,3 +1,4 @@
using System.Buffers;
using System.Net.Sockets;
namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus;
@@ -150,14 +151,17 @@ public sealed class ModbusRtuOverTcpTransport : IModbusTransport
var stream = CurrentStream;
if (stream is null) throw new InvalidOperationException("Transport not connected");
// RTU ADU: [unitId] + PDU + CRC — no MBAP header, no TxId.
var adu = ModbusRtuFraming.BuildAdu(unitId, pdu);
// RTU ADU: [unitId] + PDU + CRC — no MBAP header, no TxId. Built in a pooled buffer (R07:L5); the same bytes
// ModbusRtuFraming.BuildAdu produces (pinned by a test).
var aduLength = ModbusRtuFraming.AduLength(pdu.Length);
var adu = ArrayPool<byte>.Shared.Rent(aduLength);
ModbusRtuFraming.WriteAdu(unitId, pdu, adu.AsSpan(0, aduLength));
using var cts = CancellationTokenSource.CreateLinkedTokenSource(ct);
cts.CancelAfter(_timeout);
try
{
await stream.WriteAsync(adu.AsMemory(), cts.Token).ConfigureAwait(false);
await stream.WriteAsync(adu.AsMemory(0, aduLength), cts.Token).ConfigureAwait(false);
await stream.FlushAsync(cts.Token).ConfigureAwait(false);
// Framing owns the length-less RTU deframe + CRC + unit-address validation. A CRC-valid
@@ -181,6 +185,11 @@ public sealed class ModbusRtuOverTcpTransport : IModbusTransport
ModbusTransportDesyncException.DesyncReason.Timeout,
$"Modbus per-op response timeout after {_timeout.TotalMilliseconds:0}ms");
}
finally
{
// The write has completed, or was cancelled and the stream torn down, so nothing still reads it.
ArrayPool<byte>.Shared.Return(adu);
}
}
/// <summary>Asynchronously disposes the transport and underlying socket resources.</summary>
@@ -1,3 +1,4 @@
using System.Buffers;
using System.Net.Sockets;
namespace ZB.MOM.WW.OtOpcUa.Driver.Modbus;
@@ -59,6 +60,10 @@ public sealed class ModbusTcpTransport : IModbusTransport
/// <summary>The socket lifecycle this transport composes (test seam for its clock).</summary>
internal ModbusSocketLifecycle Lifecycle => _life;
// R07:L5: the 7-byte MBAP header of the reply being read. SendOnceAsync runs only under _gate (single-flight), so
// one buffer per transport serves every transaction (it used to be a fresh array per reply).
private readonly byte[] _header = new byte[7];
/// <inheritdoc />
public Task ConnectAsync(CancellationToken ct) => _life.ConnectAsync(ct);
@@ -139,11 +144,16 @@ public sealed class ModbusTcpTransport : IModbusTransport
if (stream is null) throw new InvalidOperationException("Transport not connected");
var txId = ++_nextTx;
// MBAP: [TxId(2)][Proto=0(2)][Length(2)][UnitId(1)] + PDU
var adu = new byte[7 + pdu.Length];
// MBAP: [TxId(2)][Proto=0(2)][Length(2)][UnitId(1)] + PDU. Built in a pooled buffer (R07:L5): the request
// never escapes, and the write has completed (or been cancelled and the socket torn down) before the finally
// hands it back. Only the reply PDU, which the caller keeps, is allocated.
// (One async method on purpose: a second awaited helper costs a state-machine box per call, which ate the saving.)
var aduLength = 7 + pdu.Length;
var adu = ArrayPool<byte>.Shared.Rent(aduLength);
adu[0] = (byte)(txId >> 8);
adu[1] = (byte)(txId & 0xFF);
// protocol id already zero
adu[2] = 0; // protocol id
adu[3] = 0;
var len = (ushort)(1 + pdu.Length); // unit id + pdu
adu[4] = (byte)(len >> 8);
adu[5] = (byte)(len & 0xFF);
@@ -154,10 +164,10 @@ public sealed class ModbusTcpTransport : IModbusTransport
cts.CancelAfter(_timeout);
try
{
await stream.WriteAsync(adu.AsMemory(), cts.Token).ConfigureAwait(false);
await stream.WriteAsync(adu.AsMemory(0, aduLength), cts.Token).ConfigureAwait(false);
await stream.FlushAsync(cts.Token).ConfigureAwait(false);
var header = new byte[7];
var header = _header;
await ReadExactlyAsync(stream, header, cts.Token).ConfigureAwait(false);
var respTxId = (ushort)((header[0] << 8) | header[1]);
if (respTxId != txId)
@@ -211,6 +221,11 @@ public sealed class ModbusTcpTransport : IModbusTransport
ModbusTransportDesyncException.DesyncReason.Timeout,
$"Modbus per-op response timeout after {_timeout.TotalMilliseconds:0}ms");
}
finally
{
// The write has completed, or was cancelled and the socket torn down, so nothing still reads it.
ArrayPool<byte>.Shared.Return(adu);
}
}
private static async Task ReadExactlyAsync(Stream s, byte[] buf, CancellationToken ct)