feat(ab-transport): Forward Close and per-exchange round-trip gate diagnostics (T21 review r1 M8/M9)

CipConnection gains two process-wide gate-diagnostic counters next to ForwardOpenCount: ForwardClosesSent (incremented as the request goes on the wire) and ForwardClosesAcked (incremented when the reply arrives within the 1 s budget).

It also records round-trip statistics for completed exchanges: ResetExchangeStats() clears them and ReadExchangeStats() returns count, min, max and mean. The cost is two Stopwatch reads and a few interlocked operations per request. No behaviour changes.

Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
2026-10-07 20:16:33 -04:00
parent 82a0c298ce
commit c90b1a8214
@@ -1,3 +1,4 @@
using System.Diagnostics;
using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip;
namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip;
@@ -23,9 +24,55 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable
/// <summary>Forward Open frames sent by every connection in the process (fallback included). Gate diagnostics.</summary>
internal static int ForwardOpenCount;
/// <summary>Forward Close requests put on the wire by every connection in the process. Gate diagnostics.</summary>
internal static int ForwardClosesSent;
/// <summary>Forward Close requests that got a reply within their budget. Gate diagnostics.</summary>
internal static int ForwardClosesAcked;
/// <summary>Channel sends by every connection in the process (keep-alives included). Gate diagnostics.</summary>
internal static int RequestsSent;
// Round-trip statistics of completed exchanges, process-wide (gate diagnostics: T21 latency attribution).
// Two Stopwatch reads and four interlocked ops per request; reset by the harness between timed passes.
private static long _exchangeCount;
private static long _exchangeTotalTicks;
private static long _exchangeMinTicks = long.MaxValue;
private static long _exchangeMaxTicks;
/// <summary>Round-trip statistics of the exchanges completed since the last <see cref="ResetExchangeStats"/>.</summary>
internal readonly record struct ExchangeStats(long Count, double MinMs, double MaxMs, double MeanMs);
/// <summary>Clears the process-wide round-trip statistics. Gate diagnostics.</summary>
internal static void ResetExchangeStats()
{
Interlocked.Exchange(ref _exchangeCount, 0);
Interlocked.Exchange(ref _exchangeTotalTicks, 0);
Interlocked.Exchange(ref _exchangeMinTicks, long.MaxValue);
Interlocked.Exchange(ref _exchangeMaxTicks, 0);
}
/// <summary>Snapshot of the process-wide round-trip statistics. Gate diagnostics.</summary>
internal static ExchangeStats ReadExchangeStats()
{
var count = Interlocked.Read(ref _exchangeCount);
if (count == 0) return new ExchangeStats(0, 0, 0, 0);
static double Ms(long ticks) => ticks * 1000.0 / Stopwatch.Frequency;
return new ExchangeStats(count, Ms(Interlocked.Read(ref _exchangeMinTicks)), Ms(Interlocked.Read(ref _exchangeMaxTicks)),
Ms(Interlocked.Read(ref _exchangeTotalTicks)) / count);
}
private static void RecordExchange(long ticks)
{
Interlocked.Increment(ref _exchangeCount);
Interlocked.Add(ref _exchangeTotalTicks, ticks);
long seen;
while (ticks < (seen = Interlocked.Read(ref _exchangeMinTicks))
&& Interlocked.CompareExchange(ref _exchangeMinTicks, ticks, seen) != seen) { }
while (ticks > (seen = Interlocked.Read(ref _exchangeMaxTicks))
&& Interlocked.CompareExchange(ref _exchangeMaxTicks, ticks, seen) != seen) { }
}
/// <summary>Open attempts (TCP connect + RegisterSession + Forward Open, a fallback open included) started by
/// every connection in the process — counted before any I/O, so a refused attempt counts too. Gate diagnostics.</summary>
internal static int OpenAttempts;
@@ -246,14 +293,17 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable
Interlocked.Exchange(ref _lastSendTimestamp, _time.GetTimestamp());
using var deadline = new CancellationTokenSource(ExchangeBudget, _time);
using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, deadline.Token);
var started = Stopwatch.GetTimestamp();
try
{
return _o.Routing switch
var reply = _o.Routing switch
{
CipRouting.Connected => await open.Client.SendUnitDataAsync(open.Conn!.OTConnectionId, cip, _o.Timeout, linked.Token).ConfigureAwait(false),
CipRouting.UnconnectedSend => await open.Client.SendRRDataAsync(ConnectionManagerServices.UnconnectedSend(cip.Span, _route), _o.Timeout, linked.Token).ConfigureAwait(false),
_ => await open.Client.SendRRDataAsync(cip, _o.Timeout, linked.Token).ConfigureAwait(false),
};
RecordExchange(Stopwatch.GetTimestamp() - started);
return reply;
}
catch (OperationCanceledException) when (!ct.IsCancellationRequested && deadline.IsCancellationRequested)
{
@@ -346,9 +396,11 @@ public sealed class CipConnection : ICipChannel, IAsyncDisposable
{
var budget = TimeSpan.FromSeconds(1);
using var cts = new CancellationTokenSource(budget, _time);
Interlocked.Increment(ref ForwardClosesSent);
await open.Client.SendRRDataAsync(
ConnectionManagerServices.ForwardClose(open.ConnectionSerial, _o.VendorId, _o.OriginatorSerial, _route),
budget, cts.Token).ConfigureAwait(false);
Interlocked.Increment(ref ForwardClosesAcked);
}
catch (Exception ex) when (ex is AbTransportException or OperationCanceledException)
{