From c90b1a8214888e71d8e2fdbc6a00827b70622b77 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 7 Oct 2026 20:16:33 -0400 Subject: [PATCH] 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 --- .../Cip/CipConnection.cs | 54 ++++++++++++++++++- 1 file changed, 53 insertions(+), 1 deletion(-) diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs index c9c87752..5004d731 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs @@ -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 /// Forward Open frames sent by every connection in the process (fallback included). Gate diagnostics. internal static int ForwardOpenCount; + /// Forward Close requests put on the wire by every connection in the process. Gate diagnostics. + internal static int ForwardClosesSent; + + /// Forward Close requests that got a reply within their budget. Gate diagnostics. + internal static int ForwardClosesAcked; + /// Channel sends by every connection in the process (keep-alives included). Gate diagnostics. 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; + + /// Round-trip statistics of the exchanges completed since the last . + internal readonly record struct ExchangeStats(long Count, double MinMs, double MaxMs, double MeanMs); + + /// Clears the process-wide round-trip statistics. Gate diagnostics. + 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); + } + + /// Snapshot of the process-wide round-trip statistics. Gate diagnostics. + 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) { } + } + /// 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. 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) {