From b82fffd2691ecf0582773a578c869606748f5bf3 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Wed, 7 Oct 2026 16:35:11 -0400 Subject: [PATCH] feat(ab-transport): per-device connection with routing modes, shared scheduler, keep-alive and Forward Open throttle --- .../Cip/CipConnection.cs | 287 +++++++++++++ .../Cip/ConnectionOptions.cs | 100 +++++ .../Cip/DeviceKey.cs | 22 + .../Cip/ForwardOpenThrottle.cs | 55 +++ .../DeviceSessionPool.cs | 129 ++++++ ....MOM.WW.OtOpcUa.Driver.Ab.Transport.csproj | 3 + .../Cip/CipConnectionTests.cs | 385 ++++++++++++++++++ .../Cip/ForwardOpenThrottleTests.cs | 156 +++++++ .../DeviceSessionPoolTests.cs | 173 ++++++++ .../Fakes/FakeCipTarget.cs | 123 ++++++ .../Fakes/ManualTimeProvider.cs | 102 +++++ 11 files changed, 1535 insertions(+) create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionOptions.cs create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/DeviceKey.cs create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ForwardOpenThrottle.cs create mode 100644 src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/DeviceSessionPool.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/CipConnectionTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipTarget.cs create mode 100644 tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/ManualTimeProvider.cs 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 new file mode 100644 index 00000000..a5443512 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/CipConnection.cs @@ -0,0 +1,287 @@ +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip; +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; + +/// +/// One device's CIP channel: one EtherNet/IP session and, when , one +/// class-3 connection. Opens lazily on first use, reconnects (through the ) +/// after the session faults, and keeps an idle connection alive. +/// +/// +/// Reconnect. Only a faulted session (timeout, I/O error, malformed frame) is reopened, and +/// only by the next send; the failing call throws its own exception. Caller cancellation never +/// faults the session (B2), so it never causes a reconnect. +/// Bounded. Every is bounded even if the caller passes no token +/// (ruling R9-3): opening (including the throttle wait and any wait behind another opener) is bounded by +/// MinForwardOpenInterval + 4 × Timeout (connect, RegisterSession, Forward Open, fallback Forward +/// Open), and the exchange by 2 × Timeout (waiting behind one in-flight exchange, then its own). +/// Exceeding either surfaces as (). +/// Connection identity. Every Forward Open attempt uses a fresh connection serial and T→O id, +/// so a reopen is never mistaken by the target for a duplicate of a connection it has not yet expired. +/// +public sealed class CipConnection : ICipChannel, IAsyncDisposable +{ + /// Forward Open frames sent by every connection in the process (fallback included). Gate diagnostics. + internal static int ForwardOpenCount; + + /// Channel sends by every connection in the process (keep-alives included). Gate diagnostics. + internal static int RequestsSent; + + /// Get Attribute Single, Identity (class 1) instance 1 attribute 1 (Vendor ID). + internal static readonly byte[] KeepAliveRequest = [0x0E, 0x03, 0x20, 0x01, 0x24, 0x01, 0x30, 0x01]; + + /// Largest unconnected message a target is guaranteed to accept. + private const int UnconnectedMessageLimit = 504; + + private sealed record Open(EncapsulationClient Client, ForwardOpenReply? Conn, ushort ConnectionSerial, int NegotiatedSize); + + private readonly DeviceKey _key; + private readonly ConnectionOptions _o; + private readonly ForwardOpenThrottle _throttle; + private readonly TimeProvider _time; + private readonly Task _predecessorClosed; + private readonly byte[] _route; + + // Never disposed: callers may still be queued on it when DisposeAsync runs, and Release on a + // disposed SemaphoreSlim throws. It owns no unmanaged handle unless AvailableWaitHandle is touched. + private readonly SemaphoreSlim _state = new(1, 1); + private readonly ITimer? _keepAlive; + private Open? _open; // written under _state; read lock-free via Volatile + private int _disposed; // checked under _state before any open; read lock-free elsewhere + private int _disposeStarted; + private long _lastSendTimestamp; + private int _keepAlivesSent; + private int _connectionSerial = Random.Shared.Next(1, ushort.MaxValue); + + /// The device; is the route (ignored for Direct). + /// Routing, sizes, timeouts and keep-alive settings. + /// The device's process-lifetime Forward Open throttle. + /// Clock for the throttle, deadlines and keep-alive. + /// Completes when the previous connection for the same key has + /// finished closing, so two connections to one device never overlap. + public CipConnection(DeviceKey key, ConnectionOptions options, ForwardOpenThrottle throttle, TimeProvider time, + Task? predecessorClosed = null) + { + _key = key ?? throw new ArgumentNullException(nameof(key)); + _o = options ?? throw new ArgumentNullException(nameof(options)); + _throttle = throttle ?? throw new ArgumentNullException(nameof(throttle)); + _time = time ?? throw new ArgumentNullException(nameof(time)); + _predecessorClosed = predecessorClosed ?? Task.CompletedTask; + // Direct ignores the path entirely (B3), so it is not even parsed. + _route = options.Routing == CipRouting.Direct ? [] : CipPath.EncodeRoute(key.CipPath); + _lastSendTimestamp = time.GetTimestamp(); + if (options.Routing == CipRouting.Connected) + _keepAlive = time.CreateTimer(static s => ((CipConnection)s!).OnKeepAliveTimer(), this, + options.KeepAliveInterval, Timeout.InfiniteTimeSpan); + } + + /// The options this connection was built with. + internal ConnectionOptions Options => _o; + + /// Test hook: keep-alives started (counted synchronously when the timer decides to send). + internal int KeepAlivesSent => Volatile.Read(ref _keepAlivesSent); + + private TimeSpan OpenBudget => _o.MinForwardOpenInterval + 4 * _o.Timeout; + private TimeSpan ExchangeBudget => 2 * _o.Timeout; + + public int MaxRequestBytes => _o.Routing switch + { + CipRouting.Connected => (Volatile.Read(ref _open)?.NegotiatedSize ?? Math.Min(_o.ConnectionSize, UnconnectedMessageLimit)) - 2, + CipRouting.UnconnectedSend => UnconnectedMessageLimit - (12 + _route.Length), + _ => UnconnectedMessageLimit, + }; + + public async Task> SendAsync(ReadOnlyMemory cipRequest, CancellationToken cancellationToken) + { + ThrowIfDisposed(); + var open = await EnsureOpenAsync(cancellationToken).ConfigureAwait(false); + return await ExchangeAsync(open, cipRequest, cancellationToken).ConfigureAwait(false); + } + + /// Returns the live (client, connection) pair, opening one if needed. The pair is returned + /// rather than re-read from fields afterwards, so a concurrent dispose can never hand the caller a + /// null (finding 16). + private async Task EnsureOpenAsync(CancellationToken ct) + { + if (Volatile.Read(ref _open) is { Client.IsConnected: true } live) return live; + + using var deadline = new CancellationTokenSource(OpenBudget, _time); + using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, deadline.Token); + try + { + await _state.WaitAsync(linked.Token).ConfigureAwait(false); + try + { + ThrowIfDisposed(); + var current = _open; + if (current is { Client.IsConnected: true }) return current; + if (current is not null) + { + Volatile.Write(ref _open, null); + await current.Client.DisposeAsync().ConfigureAwait(false); // faulted: closes without UnRegister + } + await _predecessorClosed.WaitAsync(linked.Token).ConfigureAwait(false); + var opened = await _throttle.RunAsync(OpenAsync, linked.Token).ConfigureAwait(false); + Volatile.Write(ref _open, opened); + return opened; + } + finally { _state.Release(); } + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested && deadline.IsCancellationRequested) + { + throw new AbTransportException(AbStatus.Timeout, + $"Opening {_key.Host}:{_key.Port} did not complete within {OpenBudget.TotalMilliseconds:F0} ms."); + } + } + + private async Task OpenAsync(CancellationToken ct) + { + ThrowIfDisposed(); // disposal may have begun while this attempt waited out the throttle + var client = await EncapsulationClient.ConnectAsync(_key.Host, _key.Port, _o.Timeout, ct).ConfigureAwait(false); + if (_o.Routing != CipRouting.Connected) return new Open(client, null, 0, UnconnectedMessageLimit); + try + { + var serial = NextConnectionSerial(); + var toId = (uint)Random.Shared.NextInt64(1, uint.MaxValue); + var large = _o.ConnectionSize > 511; + var size = _o.ConnectionSize; + var reply = await ForwardOpenAsync(client, large, size, serial, toId, ct).ConfigureAwait(false); + if (large && ConnectionManagerServices.ShouldFallBackToRegularOpen(reply)) + { + // Pre-v20 Logix / small-buffer targets: retry once, inside this same throttled attempt. + size = Math.Min(_o.ConnectionSize, UnconnectedMessageLimit); + reply = await ForwardOpenAsync(client, false, size, serial, toId, ct).ConfigureAwait(false); + } + return new Open(client, ConnectionManagerServices.ParseForwardOpenReply(reply), serial, size); + } + catch + { + await client.DisposeAsync().ConfigureAwait(false); + throw; + } + } + + private async Task ForwardOpenAsync(EncapsulationClient client, bool large, int size, ushort serial, + uint toId, CancellationToken ct) + { + var p = new ForwardOpenParams(large, toId, serial, _o.VendorId, _o.OriginatorSerial, _o.TimeoutMultiplier, + _o.RpiMicros, size, _route); + var frame = ConnectionManagerServices.ForwardOpen(p); + Interlocked.Increment(ref ForwardOpenCount); + return CipReply.Parse(await client.SendRRDataAsync(frame, _o.Timeout, ct).ConfigureAwait(false)); + } + + private ushort NextConnectionSerial() + { + var s = (ushort)Interlocked.Increment(ref _connectionSerial); + return s == 0 ? (ushort)Interlocked.Increment(ref _connectionSerial) : s; + } + + /// One request over an established pair, bounded by . + private async Task> ExchangeAsync(Open open, ReadOnlyMemory cip, CancellationToken ct) + { + Interlocked.Increment(ref RequestsSent); + Interlocked.Exchange(ref _lastSendTimestamp, _time.GetTimestamp()); + using var deadline = new CancellationTokenSource(ExchangeBudget, _time); + using var linked = CancellationTokenSource.CreateLinkedTokenSource(ct, deadline.Token); + try + { + return _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), + }; + } + catch (OperationCanceledException) when (!ct.IsCancellationRequested && deadline.IsCancellationRequested) + { + throw new AbTransportException(AbStatus.Timeout, + $"Request to {_key.Host}:{_key.Port} did not complete within {ExchangeBudget.TotalMilliseconds:F0} ms."); + } + } + + /// One-shot timer, re-armed after every tick so the longest idle gap is one + /// (half the connection expiry), not two. + private void OnKeepAliveTimer() + { + if (Volatile.Read(ref _disposed) != 0) return; + var interval = _o.KeepAliveInterval; + var open = Volatile.Read(ref _open); + if (open is not { Client.IsConnected: true }) + { + // Nothing to keep alive, and a keep-alive must never be what reopens a connection. + Rearm(interval); + return; + } + var idle = _time.GetElapsedTime(Interlocked.Read(ref _lastSendTimestamp)); + if (idle < interval) + { + Rearm(interval - idle); + return; + } + Interlocked.Increment(ref _keepAlivesSent); + _ = KeepAliveAsync(open); + } + + private async Task KeepAliveAsync(Open open) + { + try { await ExchangeAsync(open, KeepAliveRequest, CancellationToken.None).ConfigureAwait(false); } + catch (Exception) + { + // A timeout or I/O failure has already faulted the session; the next real send reconnects + // through the throttle. Nothing to report here: the keep-alive has no caller. + } + finally { Rearm(_o.KeepAliveInterval); } + } + + private void Rearm(TimeSpan due) + { + if (Volatile.Read(ref _disposed) != 0) return; + try { _keepAlive?.Change(due, Timeout.InfiniteTimeSpan); } + catch (ObjectDisposedException) { /* disposed concurrently */ } + } + + private void ThrowIfDisposed() + { + if (Volatile.Read(ref _disposed) != 0) + throw new AbTransportException(AbStatus.Abort, "Connection disposed."); + } + + /// Stops the keep-alive, sends Forward Close (Connected only) and closes the session, whose + /// own dispose sends UnRegisterSession. Later sends throw . + public async ValueTask DisposeAsync() + { + if (Interlocked.Exchange(ref _disposeStarted, 1) != 0) return; + + // Flag first: every caller still queued on _state sees it the moment it gets the lock and throws + // Abort without opening, so this waits behind at most the one open already in progress (bounded + // by OpenBudget), never behind a queue of them. + Volatile.Write(ref _disposed, 1); + await _state.WaitAsync().ConfigureAwait(false); + Open? open; + try + { + open = _open; + Volatile.Write(ref _open, null); + } + finally { _state.Release(); } + + if (_keepAlive is not null) await _keepAlive.DisposeAsync().ConfigureAwait(false); + if (open is null) return; + if (open.Conn is not null && open.Client.IsConnected) + { + try + { + var budget = TimeSpan.FromSeconds(1); + using var cts = new CancellationTokenSource(budget); + await open.Client.SendRRDataAsync( + ConnectionManagerServices.ForwardClose(open.ConnectionSerial, _o.VendorId, _o.OriginatorSerial, _route), + budget, cts.Token).ConfigureAwait(false); + } + catch (Exception ex) when (ex is AbTransportException or OperationCanceledException) + { + // The target expires the connection after RPI × multiplier anyway. + } + } + await open.Client.DisposeAsync().ConfigureAwait(false); + } +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionOptions.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionOptions.cs new file mode 100644 index 00000000..a8f95831 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ConnectionOptions.cs @@ -0,0 +1,100 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; + +/// How CIP requests reach the target's message router (review finding B3; the captures decide +/// each family's mode). +public enum CipRouting +{ + /// Forward Open a class-3 connection, then SendUnitData. Logix and Micro800 (empty route). + Connected, + + /// SendRRData wrapping each request in Unconnected Send (0x52) carrying the route. + UnconnectedSend, + + /// SendRRData carrying the request bare; any configured CIP path is ignored (PCCC families). + Direct, +} + +/// Per-device connection settings. Validated on assignment so an unusable value fails at the +/// configuration seam rather than on the wire. +public sealed record ConnectionOptions +{ + /// Press safety floor: no setting may let Forward Opens to one device start closer together. + public static readonly TimeSpan MinimumForwardOpenInterval = TimeSpan.FromSeconds(1); + + /// Originator serial shared by every connection this process opens. + private static readonly uint ProcessOriginatorSerial = (uint)Random.Shared.Next(1, int.MaxValue); + + private readonly TimeSpan _timeout = TimeSpan.FromSeconds(5); + private readonly int _connectionSize = 4002; + private readonly TimeSpan _minForwardOpenInterval = MinimumForwardOpenInterval; + private readonly uint _rpiMicros = 2_000_000; + private readonly byte _timeoutMultiplier = 1; + private readonly TimeSpan? _keepAliveInterval; + + public CipRouting Routing { get; init; } = CipRouting.Connected; + + /// Pack queued requests into Multiple Service Packets (the scheduler's switch). + public bool AllowPacking { get; init; } + + /// Bound on each single exchange (TCP connect, RegisterSession, Forward Open, request). + /// Must be finite and positive (ruling R9-2): an infinite timeout would let an abandoned exchange + /// hold the session's single-flight gate forever. + public TimeSpan Timeout + { + get => _timeout; + init => _timeout = value > TimeSpan.Zero && value != System.Threading.Timeout.InfiniteTimeSpan + ? value + : throw new ArgumentOutOfRangeException(nameof(Timeout), value, "Timeout must be finite and positive."); + } + + /// Requested connection size in bytes. Above 511 the open is a Large Forward Open (0x5B). + public int ConnectionSize + { + get => _connectionSize; + init => _connectionSize = value is >= 1 and <= ushort.MaxValue + ? value + : throw new ArgumentOutOfRangeException(nameof(ConnectionSize), value, "Connection size must be 1..65535."); + } + + /// Minimum time between Forward Open attempt starts for one device (≥ 1 s, press safety). + public TimeSpan MinForwardOpenInterval + { + get => _minForwardOpenInterval; + init => _minForwardOpenInterval = value >= MinimumForwardOpenInterval && value != System.Threading.Timeout.InfiniteTimeSpan + ? value + : throw new ArgumentOutOfRangeException(nameof(MinForwardOpenInterval), value, "Must be finite and at least 1 s (press safety)."); + } + + /// Requested packet interval for both directions, in microseconds. + public uint RpiMicros + { + get => _rpiMicros; + init => _rpiMicros = value > 0 ? value : throw new ArgumentOutOfRangeException(nameof(RpiMicros), value, "RPI must be positive."); + } + + /// Encoded connection timeout multiplier: the connection expires after RPI × (4 << value). + /// The default 1 means ×8, so 2 s × 8 = 16 s. + public byte TimeoutMultiplier + { + get => _timeoutMultiplier; + init => _timeoutMultiplier = value <= 7 ? value : throw new ArgumentOutOfRangeException(nameof(TimeoutMultiplier), value, "Multiplier must be 0..7."); + } + + /// How long an idle connected session waits before the target would drop it. + public TimeSpan ConnectionExpiry => TimeSpan.FromTicks(RpiMicros * 10L * (4L << TimeoutMultiplier)); + + /// Idle time after which a keep-alive is sent (Connected only). Defaults to half the + /// connection expiry: RPI × 8 / 2 = 8 s. + public TimeSpan KeepAliveInterval + { + get => _keepAliveInterval ?? ConnectionExpiry / 2; + init => _keepAliveInterval = value > TimeSpan.Zero && value != System.Threading.Timeout.InfiniteTimeSpan + ? value + : throw new ArgumentOutOfRangeException(nameof(KeepAliveInterval), value, "Keep-alive interval must be finite and positive."); + } + + public ushort VendorId { get; init; } = 0x1337; + + /// Random per process; together with the vendor id and connection serial it names a connection. + public uint OriginatorSerial { get; init; } = ProcessOriginatorSerial; +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/DeviceKey.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/DeviceKey.cs new file mode 100644 index 00000000..f0498555 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/DeviceKey.cs @@ -0,0 +1,22 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; + +/// +/// Identity of one physical CIP connection: exactly one exists per key +/// (press safety). and compare case-insensitively; +/// and compare exactly. +/// +public sealed record DeviceKey(string Host, int Port, string CipPath, string PlcAttribute) +{ + public bool Equals(DeviceKey? other) => + other is not null + && string.Equals(Host, other.Host, StringComparison.OrdinalIgnoreCase) + && Port == other.Port + && string.Equals(CipPath, other.CipPath, StringComparison.Ordinal) + && string.Equals(PlcAttribute, other.PlcAttribute, StringComparison.OrdinalIgnoreCase); + + public override int GetHashCode() => HashCode.Combine( + StringComparer.OrdinalIgnoreCase.GetHashCode(Host), + Port, + StringComparer.Ordinal.GetHashCode(CipPath), + StringComparer.OrdinalIgnoreCase.GetHashCode(PlcAttribute)); +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ForwardOpenThrottle.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ForwardOpenThrottle.cs new file mode 100644 index 00000000..f25b24a5 --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/Cip/ForwardOpenThrottle.cs @@ -0,0 +1,55 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; + +/// Press safety: at most one Forward Open attempt in flight per device and a minimum +/// interval between attempt starts. A misconfigured client once drove ~15 Forward Opens/s at a +/// production press (ignitionoee vite-press-oee-bind.md, 2026-09-23); this is what prevents it. +/// One instance per , kept by for the +/// process lifetime so a reconnect after the last lease was released is still limited. Its semaphore is +/// therefore never disposed. The interval is measured from the previous attempt's start, whatever +/// that attempt's outcome. +public sealed class ForwardOpenThrottle +{ + private readonly SemaphoreSlim _inFlight = new(1, 1); + private readonly TimeSpan _minInterval; + private readonly TimeProvider _time; + private DateTimeOffset? _lastStart; // only touched while holding _inFlight + + public ForwardOpenThrottle(TimeSpan minInterval, TimeProvider time) + { + if (minInterval < TimeSpan.Zero || minInterval == Timeout.InfiniteTimeSpan) + throw new ArgumentOutOfRangeException(nameof(minInterval), minInterval, "Interval must be finite and non-negative."); + _minInterval = minInterval; + _time = time ?? throw new ArgumentNullException(nameof(time)); + } + + /// Run one open attempt once the throttle allows it. Throws + /// () when + /// fires before the attempt could start; the attempt itself sees too. + public async Task RunAsync(Func> open, CancellationToken ct) + { + ArgumentNullException.ThrowIfNull(open); + try { await _inFlight.WaitAsync(ct).ConfigureAwait(false); } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { throw Throttled("another attempt is in flight"); } + + try + { + if (_lastStart is { } last) + { + var wait = last + _minInterval - _time.GetUtcNow(); + if (wait > TimeSpan.Zero) + { + try { await Task.Delay(wait, _time, ct).ConfigureAwait(false); } + catch (OperationCanceledException) when (ct.IsCancellationRequested) + { throw Throttled("the minimum interval had not elapsed"); } + } + } + _lastStart = _time.GetUtcNow(); + return await open(ct).ConfigureAwait(false); + } + finally { _inFlight.Release(); } + } + + private static AbTransportException Throttled(string why) => + new(AbStatus.BadConnection, $"Forward Open throttled; caller deadline reached before {why}."); +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/DeviceSessionPool.cs b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/DeviceSessionPool.cs new file mode 100644 index 00000000..b254f80c --- /dev/null +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/DeviceSessionPool.cs @@ -0,0 +1,129 @@ +using Microsoft.Extensions.Logging; +using Microsoft.Extensions.Logging.Abstractions; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport; + +/// +/// Process-wide owner of device connections: exactly one and one +/// per (review B1), handed out as ref-counted +/// s. The last lease's dispose closes the connection; a later acquire builds +/// fresh instances, which do not open until the previous connection has finished closing. +/// +/// Forward Open throttles are kept per key for the process lifetime and never pruned, so a +/// reconnect after the last release is still rate-limited against the previous attempt. +public sealed class DeviceSessionPool +{ + /// The process-wide pool the drivers use. Tests build their own. + public static DeviceSessionPool Shared { get; } = new(); + + internal sealed class Entry(CipConnection connection, RequestScheduler scheduler, ConnectionOptions options) + { + public CipConnection Connection { get; } = connection; + public RequestScheduler Scheduler { get; } = scheduler; + public ConnectionOptions Options { get; } = options; + public int RefCount; // under the pool lock + public bool MismatchLogged; // under the pool lock + } + + private readonly object _lock = new(); + private readonly Dictionary _entries = new(); + private readonly Dictionary _throttles = new(); + private readonly Dictionary _closing = new(); + private readonly TimeProvider _time; + private readonly ILogger _logger; + + public DeviceSessionPool(TimeProvider? time = null, ILogger? logger = null) + { + _time = time ?? TimeProvider.System; + _logger = logger ?? NullLogger.Instance; + } + + /// Test hook: keys with a live entry. + internal int EntryCount { get { lock (_lock) return _entries.Count; } } + + /// Lease the shared connection for , creating it on first use. Performs + /// no I/O: the connection opens on its first request. When the key already has an entry, that entry's + /// options win; a differing is logged once per entry. + public ConnectionLease Acquire(DeviceKey key, ConnectionOptions options) + { + ArgumentNullException.ThrowIfNull(key); + ArgumentNullException.ThrowIfNull(options); + lock (_lock) + { + if (!_entries.TryGetValue(key, out var entry)) + { + if (!_throttles.TryGetValue(key, out var throttle)) + _throttles[key] = throttle = new ForwardOpenThrottle(options.MinForwardOpenInterval, _time); + var predecessor = _closing.GetValueOrDefault(key); + var connection = new CipConnection(key, options, throttle, _time, predecessor); + entry = new Entry(connection, new RequestScheduler(connection, options.AllowPacking), options); + _entries[key] = entry; + } + else if (!entry.Options.Equals(options) && !entry.MismatchLogged) + { + entry.MismatchLogged = true; + _logger.LogWarning( + "Device {Host}:{Port} path '{CipPath}' is already connected with different connection options; " + + "the existing options ({Existing}) stay in force and the requested ({Requested}) are ignored.", + key.Host, key.Port, key.CipPath, entry.Options, options); + } + entry.RefCount++; + return new ConnectionLease(this, key, entry); + } + } + + internal async ValueTask ReleaseAsync(DeviceKey key, Entry entry) + { + TaskCompletionSource closed; + lock (_lock) + { + if (--entry.RefCount > 0) return; + _entries.Remove(key); + closed = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + _closing[key] = closed.Task; + } + + // Outside the lock: scheduler first (fails its waiters, stops sending), then Forward Close + + // UnRegisterSession via the connection. + try + { + await entry.Scheduler.DisposeAsync().ConfigureAwait(false); + await entry.Connection.DisposeAsync().ConfigureAwait(false); + } + finally + { + lock (_lock) + { + if (_closing.TryGetValue(key, out var t) && t == closed.Task) _closing.Remove(key); + } + closed.TrySetResult(); // never faults: a successor only needs to know the close is over + } + } +} + +/// A reference on a device's shared connection. Dispose exactly once (further disposes are no-ops). +public sealed class ConnectionLease : IAsyncDisposable +{ + private readonly DeviceSessionPool _pool; + private readonly DeviceKey _key; + private readonly DeviceSessionPool.Entry _entry; + private int _disposed; + + internal ConnectionLease(DeviceSessionPool pool, DeviceKey key, DeviceSessionPool.Entry entry) + { + _pool = pool; + _key = key; + _entry = entry; + } + + public DeviceKey Key => _key; + + /// The device's channel. Prefer : it serializes and packs requests. + public ICipChannel Channel => _entry.Connection; + + /// The device's single request scheduler, shared by every lease on the key. + public RequestScheduler Scheduler => _entry.Scheduler; + + public ValueTask DisposeAsync() => + Interlocked.Exchange(ref _disposed, 1) == 0 ? _pool.ReleaseAsync(_key, _entry) : ValueTask.CompletedTask; +} diff --git a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.csproj b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.csproj index 2f48e07a..c15cdd13 100644 --- a/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.csproj +++ b/src/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.csproj @@ -15,5 +15,8 @@ + + + diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/CipConnectionTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/CipConnectionTests.cs new file mode 100644 index 00000000..84c25f02 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/CipConnectionTests.cs @@ -0,0 +1,385 @@ +using System.Buffers.Binary; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes; +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Cip; + +/// Serializes every test class that asserts on 's process-wide +/// diagnostic counters, so no parallel test can move them mid-assertion. +[CollectionDefinition(Name, DisableParallelization = true)] +public sealed class CipConnectionStatics +{ + public const string Name = "CipConnection statics"; +} + +/// against . Wire I/O is real loopback; +/// the throttle, deadlines and keep-alive run on . +[Collection(CipConnectionStatics.Name)] +public sealed class CipConnectionTests +{ + private static readonly TimeSpan Guard = TimeSpan.FromSeconds(10); + private static readonly byte[] ReadTag = [0x4C, 0x02, 0x91, 0x03, 0x54, 0x61, 0x67, 0x00, 0x01, 0x00]; + private static CancellationToken TestCt => TestContext.Current.CancellationToken; + + private static CipConnection Create(FakeCipTarget target, ConnectionOptions o, ManualTimeProvider time, string path = "1,0") => + new(new DeviceKey("127.0.0.1", target.Port, path, "ControlLogix"), o, + new ForwardOpenThrottle(o.MinForwardOpenInterval, time), time); + + private static ConnectionOptions Options(CipRouting routing = CipRouting.Connected, int size = 4002, double timeoutSeconds = 2) => + new() { Routing = routing, ConnectionSize = size, Timeout = TimeSpan.FromSeconds(timeoutSeconds) }; + + private static ushort RegularOTParams(byte[] fo) => BinaryPrimitives.ReadUInt16LittleEndian(fo.AsSpan(32)); + + // ---- options --------------------------------------------------------------------------------- + + [Fact] + public void Defaults_match_the_press_safety_values() + { + var o = new ConnectionOptions(); + o.MinForwardOpenInterval.ShouldBe(TimeSpan.FromSeconds(1)); + o.RpiMicros.ShouldBe(2_000_000u); + o.TimeoutMultiplier.ShouldBe((byte)1); + o.ConnectionExpiry.ShouldBe(TimeSpan.FromSeconds(16)); + o.KeepAliveInterval.ShouldBe(TimeSpan.FromSeconds(8)); + o.VendorId.ShouldBe((ushort)0x1337); + new ConnectionOptions().OriginatorSerial.ShouldBe(o.OriginatorSerial); // per process, not per instance + } + + public static TheoryData UnusableTimeouts => new() { Timeout.InfiniteTimeSpan, TimeSpan.Zero, TimeSpan.FromSeconds(-1) }; + + [Theory] + [MemberData(nameof(UnusableTimeouts))] + public void Options_reject_a_non_finite_or_non_positive_timeout(TimeSpan timeout) => + Should.Throw(() => new ConnectionOptions { Timeout = timeout }); + + [Fact] + public void Options_reject_a_forward_open_interval_below_one_second() => + Should.Throw(() => new ConnectionOptions { MinForwardOpenInterval = TimeSpan.FromMilliseconds(999) }); + + // ---- Large Forward Open fallback ------------------------------------------------------------- + + [Fact] + public async Task Large_open_service_not_supported_falls_back_to_regular_open_once() + { + await using var target = new FakeCipTarget(); + target.ForwardOpenReply = fo => fo[0] == 0x5B ? FakeCipTarget.Refuse(0x08)(fo) : FakeCipTarget.Accept(fo); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(), time); + var before = Volatile.Read(ref CipConnection.ForwardOpenCount); + + await conn.SendAsync(ReadTag, TestCt); + + var fos = target.ForwardOpenRequests; + fos.Count.ShouldBe(2); + fos[0][0].ShouldBe((byte)0x5B); + fos[1][0].ShouldBe((byte)0x54); + RegularOTParams(fos[1]).ShouldBe((ushort)(0x4200 | 504)); + conn.MaxRequestBytes.ShouldBe(502); + (Volatile.Read(ref CipConnection.ForwardOpenCount) - before).ShouldBe(2); + target.ConnectedRequests.ShouldBe(1); + } + + [Fact] + public async Task Large_open_bad_size_falls_back_to_regular_open_once() + { + await using var target = new FakeCipTarget(); + target.ForwardOpenReply = fo => fo[0] == 0x5B ? FakeCipTarget.Refuse(0x01, 0x0109)(fo) : FakeCipTarget.Accept(fo); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(), time); + + await conn.SendAsync(ReadTag, TestCt); + + target.ForwardOpenRequests.Select(f => f[0]).ShouldBe(new byte[] { 0x5B, 0x54 }); + RegularOTParams(target.ForwardOpenRequests[1]).ShouldBe((ushort)(0x4200 | 504)); + } + + [Fact] + public async Task A_refused_fallback_is_not_retried_again() + { + await using var target = new FakeCipTarget { ForwardOpenReply = FakeCipTarget.Refuse(0x08) }; + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(), time); + + await Should.ThrowAsync(() => conn.SendAsync(ReadTag, TestCt)); + + target.ForwardOpens.ShouldBe(2); // Large, then exactly one regular + } + + [Fact] + public async Task Connection_in_use_does_not_fall_back() + { + await using var target = new FakeCipTarget { ForwardOpenReply = FakeCipTarget.Refuse(0x01, 0x0100) }; + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(), time); + + var ex = await Should.ThrowAsync(() => conn.SendAsync(ReadTag, TestCt)); + + ex.Message.ShouldContain("Forward Open refused"); + target.ForwardOpens.ShouldBe(1); + target.ConnectedRequests.ShouldBe(0); + } + + [Fact] + public async Task A_small_connection_size_uses_a_regular_open() + { + await using var target = new FakeCipTarget(); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(size: 504), time); + + await conn.SendAsync(ReadTag, TestCt); + + target.ForwardOpenRequests.Single()[0].ShouldBe((byte)0x54); + conn.MaxRequestBytes.ShouldBe(502); + } + + // ---- routing (B3) ---------------------------------------------------------------------------- + + [Fact] + public async Task Direct_routing_sends_bare_cip_even_with_a_path() + { + await using var target = new FakeCipTarget(); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(CipRouting.Direct), time, path: "1,0"); + + var reply = await conn.SendAsync(ReadTag, TestCt); + + CipReply.Parse(reply).GeneralStatus.ShouldBe((byte)0); + target.DirectCips.Single().ShouldBe(ReadTag); + target.ForwardOpens.ShouldBe(0); + target.UnconnectedSends.ShouldBe(0); + target.ConnectedRequests.ShouldBe(0); + conn.MaxRequestBytes.ShouldBe(504); + } + + [Fact] + public async Task UnconnectedSend_routing_wraps_in_0x52() + { + await using var target = new FakeCipTarget(); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(CipRouting.UnconnectedSend), time, path: "1,0"); + + await conn.SendAsync(ReadTag, TestCt); + + target.UnconnectedEmbedded.Single().ShouldBe(ReadTag); + var raw = FakeEipServer.RequestCip(target.Server.Received.Last(f => FakeEipServer.Command(f) == 0x006F)); + raw.ShouldBe(ConnectionManagerServices.UnconnectedSend(ReadTag, new byte[] { 0x01, 0x00 })); + target.ForwardOpens.ShouldBe(0); + target.DirectRequests.ShouldBe(0); + conn.MaxRequestBytes.ShouldBe(504 - (12 + 2)); + } + + [Fact] + public async Task Connected_routing_never_sends_bare() + { + await using var target = new FakeCipTarget(); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(), time); + + for (var i = 0; i < 3; i++) await conn.SendAsync(ReadTag, TestCt); + + target.ConnectedCips.Count.ShouldBe(3); + target.ConnectedCips.ShouldAllBe(c => c.SequenceEqual(ReadTag)); + target.ForwardOpens.ShouldBe(1); + target.DirectRequests.ShouldBe(0); + target.UnconnectedSends.ShouldBe(0); + conn.MaxRequestBytes.ShouldBe(4000); + } + + [Fact] + public async Task Connected_routing_with_an_empty_route_opens_straight_to_the_message_router() + { + // R9-1: Micro800 is Connected with an empty route; libplctag's Forward Open path is just 20 02 24 01. + await using var target = new FakeCipTarget(); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(size: 504), time, path: ""); + + await conn.SendAsync(ReadTag, TestCt); + + var fo = target.ForwardOpenRequests.Single(); + fo[^6..].ShouldBe(new byte[] { 0xA3, 0x02, 0x20, 0x02, 0x24, 0x01 }); + target.ConnectedRequests.ShouldBe(1); + } + + // ---- reconnect ------------------------------------------------------------------------------- + + [Fact] + public async Task Caller_cancellation_does_not_reconnect() + { + await using var target = new FakeCipTarget(); + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + var held = 0; + target.ConnectedHandler = async cip => + { + if (Interlocked.Increment(ref held) == 1) await release.Task; + return [(byte)(cip[0] | 0x80), 0, 0, 0]; + }; + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(), time); + var before = Volatile.Read(ref CipConnection.ForwardOpenCount); + + using var cts = CancellationTokenSource.CreateLinkedTokenSource(TestCt); + var pending = conn.SendAsync(ReadTag, cts.Token); + await FakeCipTarget.Until(() => target.ConnectedRequests == 1, Guard, "the connected request"); + await cts.CancelAsync(); + await Should.ThrowAsync(() => pending); + release.SetResult(); + + await conn.SendAsync(ReadTag, TestCt); + + (Volatile.Read(ref CipConnection.ForwardOpenCount) - before).ShouldBe(1); + target.ForwardOpens.ShouldBe(1); + target.Server.Connections.ShouldBe(1); + target.ConnectedRequests.ShouldBe(2); + } + + [Fact] + public async Task Timeout_reconnects_through_the_throttle() + { + await using var target = new FakeCipTarget(); + var answer = 1; + target.ConnectedHandler = cip => Task.FromResult( + Volatile.Read(ref answer) == 1 ? [(byte)(cip[0] | 0x80), 0, 0, 0] : null); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(timeoutSeconds: 0.3), time); + + await conn.SendAsync(ReadTag, TestCt); // first Forward Open, at fake t0 + Volatile.Write(ref answer, 0); + var timedOut = await Should.ThrowAsync(() => conn.SendAsync(ReadTag, TestCt)); + timedOut.Status.ShouldBe(AbStatus.Timeout); // the failing call keeps its own error + target.ForwardOpens.ShouldBe(1); + + Volatile.Write(ref answer, 1); + var reconnect = conn.SendAsync(ReadTag, TestCt); + await time.WaitForTimerDueAtAsync(time.Start + TimeSpan.FromSeconds(1), Guard); // parked in the throttle + target.ForwardOpens.ShouldBe(1); + reconnect.IsCompleted.ShouldBeFalse(); + + time.Advance(TimeSpan.FromSeconds(1)); + await reconnect.WaitAsync(Guard, TestCt); + + target.ForwardOpens.ShouldBe(2); + target.Server.Connections.ShouldBe(2); + // A fresh connection serial per attempt, so the target never sees a duplicate of the stale connection. + var serials = target.ForwardOpenRequests.Select(f => BinaryPrimitives.ReadUInt16LittleEndian(f.AsSpan(16))).ToArray(); + serials[1].ShouldNotBe(serials[0]); + } + + [Fact] + public async Task A_stalled_open_is_bounded_by_the_deadline() + { + // R9-3: no caller token at all, and the target never answers the Forward Open. + await using var target = new FakeCipTarget(); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(timeoutSeconds: 30), time); + target.Server.Stall = true; + + var send = conn.SendAsync(ReadTag, CancellationToken.None); + await FakeCipTarget.Until(() => target.ForwardOpens == 0 && target.Server.Received.Count >= 2, Guard, "the Forward Open frame"); + var openBudget = TimeSpan.FromSeconds(1) + 4 * TimeSpan.FromSeconds(30); + await time.WaitForTimerDueAtAsync(time.Start + openBudget, Guard); + time.Advance(openBudget); + + var ex = await Should.ThrowAsync(() => send.WaitAsync(Guard, TestCt)); + ex.Status.ShouldBe(AbStatus.Timeout); + } + + // ---- keep-alive ------------------------------------------------------------------------------ + + [Fact] + public async Task Idle_connection_sends_one_keepalive_per_interval() + { + await using var target = new FakeCipTarget(); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(), time); + var before = Volatile.Read(ref CipConnection.RequestsSent); + await conn.SendAsync(ReadTag, TestCt); + + time.Advance(TimeSpan.FromSeconds(8)); + conn.KeepAlivesSent.ShouldBe(1); + await FakeCipTarget.Until(() => target.ConnectedRequests == 2, Guard, "the keep-alive"); + target.ConnectedCips[1].ShouldBe(CipConnection.KeepAliveRequest); + await time.WaitForTimerDueAtAsync(time.Start + TimeSpan.FromSeconds(16), Guard); // re-armed + + time.Advance(TimeSpan.FromSeconds(7)); + conn.KeepAlivesSent.ShouldBe(1); // exactly one per idle interval + + time.Advance(TimeSpan.FromSeconds(1)); + conn.KeepAlivesSent.ShouldBe(2); + await FakeCipTarget.Until(() => target.ConnectedRequests == 3, Guard, "the second keep-alive"); + (Volatile.Read(ref CipConnection.RequestsSent) - before).ShouldBe(3); // keep-alives count + target.ForwardOpens.ShouldBe(1); + } + + [Fact] + public async Task Traffic_suppresses_keepalive() + { + await using var target = new FakeCipTarget(); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(), time); + await conn.SendAsync(ReadTag, TestCt); + + for (var i = 0; i < 4; i++) + { + time.Advance(TimeSpan.FromSeconds(5)); + await conn.SendAsync(ReadTag, TestCt); + } + + conn.KeepAlivesSent.ShouldBe(0); + target.ConnectedCips.ShouldAllBe(c => c.SequenceEqual(ReadTag)); + target.ConnectedRequests.ShouldBe(5); + } + + [Fact] + public async Task No_keepalive_is_sent_when_the_connection_was_never_opened() + { + await using var target = new FakeCipTarget(); + var time = new ManualTimeProvider(); + await using var conn = Create(target, Options(), time); + + time.Advance(TimeSpan.FromSeconds(30)); + + conn.KeepAlivesSent.ShouldBe(0); + target.ForwardOpens.ShouldBe(0); + target.Server.Connections.ShouldBe(0); + } + + // ---- disposal -------------------------------------------------------------------------------- + + [Fact] + public async Task Send_after_dispose_throws_Abort() + { + await using var target = new FakeCipTarget(); + var time = new ManualTimeProvider(); + var conn = Create(target, Options(), time); + await conn.SendAsync(ReadTag, TestCt); + + await conn.DisposeAsync(); + + var ex = await Should.ThrowAsync(() => conn.SendAsync(ReadTag, TestCt)); + ex.Status.ShouldBe(AbStatus.Abort); + target.ForwardCloses.ShouldBe(1); + await conn.DisposeAsync(); // idempotent + } + + [Fact] + public async Task Callers_queued_behind_an_open_get_Abort_when_disposed() + { + await using var target = new FakeCipTarget { ForwardOpenReply = FakeCipTarget.Refuse(0x01, 0x0113) }; + var time = new ManualTimeProvider(); + var conn = Create(target, Options(), time); + await Should.ThrowAsync(() => conn.SendAsync(ReadTag, TestCt)); // attempt 1, t0 + + var parked = conn.SendAsync(ReadTag, TestCt); // waits in the throttle + await time.WaitForTimerDueAtAsync(time.Start + TimeSpan.FromSeconds(1), Guard); + var queued = Enumerable.Range(0, 5).Select(_ => conn.SendAsync(ReadTag, TestCt)).ToArray(); + var dispose = conn.DisposeAsync().AsTask(); + time.Advance(TimeSpan.FromSeconds(1)); + + (await Should.ThrowAsync(() => parked.WaitAsync(Guard, TestCt))).Status.ShouldBe(AbStatus.Abort); + foreach (var q in queued) + (await Should.ThrowAsync(() => q.WaitAsync(Guard, TestCt))).Status.ShouldBe(AbStatus.Abort); + await dispose.WaitAsync(Guard, TestCt); + target.ForwardOpens.ShouldBe(1); // nothing is opened once disposal has begun + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs new file mode 100644 index 00000000..bf2124de --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Cip/ForwardOpenThrottleTests.cs @@ -0,0 +1,156 @@ +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes; +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Cip; + +/// Press safety: on fake time, and the storm it exists to stop. +[Collection(CipConnectionStatics.Name)] // the storm test moves CipConnection's static counters +public sealed class ForwardOpenThrottleTests +{ + private static readonly TimeSpan Second = TimeSpan.FromSeconds(1); + private static readonly TimeSpan Guard = TimeSpan.FromSeconds(10); + private static CancellationToken TestCt => TestContext.Current.CancellationToken; + + [Fact] + public async Task Second_attempt_inside_interval_waits() + { + var time = new ManualTimeProvider(); + var throttle = new ForwardOpenThrottle(Second, time); + await throttle.RunAsync(_ => Task.FromResult(1), TestCt); + time.Advance(TimeSpan.FromMilliseconds(400)); + + var started = 0; + var second = throttle.RunAsync(_ => { Interlocked.Increment(ref started); return Task.FromResult(2); }, TestCt); + await time.WaitForTimerDueAtAsync(time.Start + Second, Guard); // waits for the remainder, measured from attempt 1's start + time.Advance(TimeSpan.FromMilliseconds(599)); + Volatile.Read(ref started).ShouldBe(0); + second.IsCompleted.ShouldBeFalse(); + + time.Advance(TimeSpan.FromMilliseconds(1)); + (await second.WaitAsync(Guard, TestCt)).ShouldBe(2); + Volatile.Read(ref started).ShouldBe(1); + } + + [Fact] + public async Task Attempt_after_interval_proceeds_immediately() + { + var time = new ManualTimeProvider(); + var throttle = new ForwardOpenThrottle(Second, time); + await throttle.RunAsync(_ => Task.FromResult(1), TestCt); + time.Advance(Second); + + var started = false; + var second = throttle.RunAsync(_ => { started = true; return Task.FromResult(2); }, TestCt); + + started.ShouldBeTrue(); // ran synchronously: no wait was needed + second.IsCompletedSuccessfully.ShouldBeTrue(); + time.ArmedTimers.ShouldBe(0); + } + + [Fact] + public async Task Only_one_attempt_in_flight() + { + var time = new ManualTimeProvider(); + var throttle = new ForwardOpenThrottle(Second, time); + var inFlight = 0; + var maxInFlight = 0; + var release = new TaskCompletionSource(TaskCreationOptions.RunContinuationsAsynchronously); + + async Task Attempt(CancellationToken _) + { + var now = Interlocked.Increment(ref inFlight); + InterlockedMax(ref maxInFlight, now); + await release.Task; + Interlocked.Decrement(ref inFlight); + return now; + } + + var first = throttle.RunAsync(Attempt, TestCt); + time.Advance(TimeSpan.FromSeconds(5)); // the interval is no obstacle; only the in-flight one is + var others = Enumerable.Range(0, 4).Select(_ => throttle.RunAsync(Attempt, TestCt)).ToArray(); + await Task.Delay(50, TestCt); + Volatile.Read(ref inFlight).ShouldBe(1); + others.ShouldAllBe(t => !t.IsCompleted); + + release.SetResult(); + await first.WaitAsync(Guard, TestCt); + // Attempt 2 starts at once (attempt 1 started 5 s ago); each later one then waits out the + // interval from its predecessor's start. + await others[0].WaitAsync(Guard, TestCt); + for (var i = 1; i < others.Length; i++) + { + await time.WaitForTimerDueAtAsync(time.GetUtcNow() + Second, Guard); + time.Advance(Second); + await FakeCipTarget.Until(() => others.Count(t => t.IsCompleted) == i + 1, Guard, $"attempt {i + 2}"); + } + Volatile.Read(ref maxInFlight).ShouldBe(1); + } + + [Fact] + public async Task Caller_deadline_while_waiting_throws_throttled() + { + var time = new ManualTimeProvider(); + var throttle = new ForwardOpenThrottle(Second, time); + await throttle.RunAsync(_ => Task.FromResult(1), TestCt); + using var cts = new CancellationTokenSource(); + var invoked = false; + + var second = throttle.RunAsync(_ => { invoked = true; return Task.FromResult(2); }, cts.Token); + await time.WaitForTimerDueAtAsync(time.Start + Second, Guard); + await cts.CancelAsync(); + + var ex = await Should.ThrowAsync(() => second.WaitAsync(Guard, TestCt)); + ex.Status.ShouldBe(AbStatus.BadConnection); + ex.Message.ShouldContain("Forward Open throttled"); + invoked.ShouldBeFalse(); + } + + [Fact] + public async Task Refusing_target_sees_at_most_one_forward_open_per_interval() + { + // The storm: 50 callers hammering a device that refuses every Forward Open, for 3 s of fake time. + // Attempt starts are >= 1 s apart, so t = 0, 1, 2, 3 is the most there can be. + await using var target = new FakeCipTarget { ForwardOpenReply = FakeCipTarget.Refuse(0x01, 0x0113) }; + var time = new ManualTimeProvider(); + var options = new ConnectionOptions { ConnectionSize = 504, Timeout = TimeSpan.FromSeconds(2) }; + var conn = new CipConnection(new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix"), options, + new ForwardOpenThrottle(options.MinForwardOpenInterval, time), time); + using var stop = CancellationTokenSource.CreateLinkedTokenSource(TestCt); + var failures = 0; + + var callers = Enumerable.Range(0, 50).Select(_ => Task.Run(async () => + { + while (!stop.IsCancellationRequested) + { + try { await conn.SendAsync(new byte[] { 0x4C, 0x00 }, stop.Token); } + catch (AbTransportException) { Interlocked.Increment(ref failures); } + catch (OperationCanceledException) { /* stopping */ } + } + }, TestCt)).ToArray(); + + for (var step = 0; step < 30; step++) + { + await Task.Delay(10, TestCt); + time.Advance(TimeSpan.FromMilliseconds(100)); + } + await Task.Delay(100, TestCt); // let any attempt released by the last step reach the wire + + await stop.CancelAsync(); + await Task.WhenAll(callers).WaitAsync(Guard, TestCt); + await conn.DisposeAsync(); + + time.GetUtcNow().ShouldBe(time.Start + TimeSpan.FromSeconds(3)); + target.ForwardOpens.ShouldBeLessThanOrEqualTo(4); + target.ForwardOpens.ShouldBeGreaterThanOrEqualTo(1); + Volatile.Read(ref failures).ShouldBeGreaterThanOrEqualTo(target.ForwardOpens); + target.ConnectedRequests.ShouldBe(0); + } + + private static void InterlockedMax(ref int location, int value) + { + int current; + while (value > (current = Volatile.Read(ref location)) && + Interlocked.CompareExchange(ref location, value, current) != current) { } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs new file mode 100644 index 00000000..065ed32c --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/DeviceSessionPoolTests.cs @@ -0,0 +1,173 @@ +using Microsoft.Extensions.Logging; +using Shouldly; +using Xunit; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Cip; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes; +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests; + +[Collection(CipConnectionStatics.Name)] +public sealed class DeviceSessionPoolTests +{ + private static readonly TimeSpan Guard = TimeSpan.FromSeconds(10); + private static readonly byte[] ReadTag = [0x4C, 0x02, 0x91, 0x03, 0x54, 0x61, 0x67, 0x00, 0x01, 0x00]; + private static CancellationToken TestCt => TestContext.Current.CancellationToken; + + private static ConnectionOptions Options(int size = 4002) => new() { ConnectionSize = size, Timeout = TimeSpan.FromSeconds(2) }; + + [Fact] + public void Device_keys_compare_host_and_plc_attribute_case_insensitively() + { + var a = new DeviceKey("Press-07.plant.local", 44818, "1,0", "ControlLogix"); + var b = new DeviceKey("PRESS-07.PLANT.LOCAL", 44818, "1,0", "controllogix"); + a.ShouldBe(b); + a.GetHashCode().ShouldBe(b.GetHashCode()); + a.ShouldNotBe(a with { CipPath = "1,1" }); + a.ShouldNotBe(a with { Port = 2222 }); + } + + [Fact] + public async Task Leases_for_one_key_share_one_scheduler() + { + var pool = new DeviceSessionPool(new ManualTimeProvider()); + var key = new DeviceKey("plc1", 44818, "1,0", "ControlLogix"); + + await using var a = pool.Acquire(key, Options()); + await using var b = pool.Acquire(key with { Host = "PLC1" }, Options()); + await using var other = pool.Acquire(key with { CipPath = "1,2" }, Options()); + + b.Scheduler.ShouldBeSameAs(a.Scheduler); + b.Channel.ShouldBeSameAs(a.Channel); + other.Scheduler.ShouldNotBeSameAs(a.Scheduler); + other.Channel.ShouldNotBeSameAs(a.Channel); + pool.EntryCount.ShouldBe(2); + } + + [Fact] + public async Task A_second_acquire_with_different_options_reuses_the_first_and_logs_once() + { + var logger = new CapturingLogger(); + var pool = new DeviceSessionPool(new ManualTimeProvider(), logger); + var key = new DeviceKey("plc1", 44818, "1,0", "ControlLogix"); + var first = Options(4002); + var second = Options(504) with { AllowPacking = true }; + + await using var a = pool.Acquire(key, first); + await using var b = pool.Acquire(key, second); + await using var c = pool.Acquire(key, second); + await using var d = pool.Acquire(key, first); + + ((CipConnection)b.Channel).Options.ShouldBe(first); + ((CipConnection)c.Channel).Options.ShouldBe(first); + logger.Warnings.ShouldBe(1); + } + + [Fact] + public async Task Last_lease_dispose_sends_forward_close_then_unregister() + { + await using var target = new FakeCipTarget(); + var pool = new DeviceSessionPool(new ManualTimeProvider()); + var key = new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix"); + var a = pool.Acquire(key, Options()); + var b = pool.Acquire(key, Options()); + (await a.Scheduler.SendAsync(ReadTag, TestCt)).GeneralStatus.ShouldBe((byte)0); + + await a.DisposeAsync(); + await a.DisposeAsync(); // idempotent: must not release b's reference + target.ForwardCloses.ShouldBe(0); + (await b.Scheduler.SendAsync(ReadTag, TestCt)).GeneralStatus.ShouldBe((byte)0); + + await b.DisposeAsync(); + await target.Server.WaitForFrameAsync(f => FakeEipServer.Command(f) == EncapCommand.UnRegisterSession, Guard); + + target.ForwardCloses.ShouldBe(1); + var frames = target.Server.Received.ToList(); + var close = frames.FindIndex(f => FakeEipServer.Command(f) == EncapCommand.SendRRData + && FakeCipTarget.IsForwardClose(FakeEipServer.RequestCip(f))); + var unregister = frames.FindIndex(f => FakeEipServer.Command(f) == EncapCommand.UnRegisterSession); + close.ShouldBeGreaterThanOrEqualTo(0); + close.ShouldBeLessThan(unregister); + // The Forward Close names the connection that was opened. + target.ForwardCloseRequests.Single().AsSpan(8, 8).ToArray() + .ShouldBe(target.ForwardOpenRequests.Single().AsSpan(16, 8).ToArray()); + pool.EntryCount.ShouldBe(0); + await Should.ThrowAsync(() => b.Scheduler.SendAsync(ReadTag, TestCt)); + } + + [Fact] + public async Task A_later_acquire_builds_new_instances() + { + var pool = new DeviceSessionPool(new ManualTimeProvider()); + var key = new DeviceKey("plc1", 44818, "1,0", "ControlLogix"); + var first = pool.Acquire(key, Options()); + var scheduler = first.Scheduler; + var channel = first.Channel; + await first.DisposeAsync(); + + await using var second = pool.Acquire(key, Options()); + + second.Scheduler.ShouldNotBeSameAs(scheduler); + second.Channel.ShouldNotBeSameAs(channel); + } + + [Fact] + public async Task A_reconnect_after_the_last_release_is_still_throttled() + { + await using var target = new FakeCipTarget(); + var time = new ManualTimeProvider(); + var pool = new DeviceSessionPool(time); + var key = new DeviceKey("127.0.0.1", target.Port, "1,0", "ControlLogix"); + + await using (var first = pool.Acquire(key, Options())) + await first.Scheduler.SendAsync(ReadTag, TestCt); // Forward Open at fake t0 + + await using var second = pool.Acquire(key, Options()); + var send = second.Scheduler.SendAsync(ReadTag, TestCt); + await time.WaitForTimerDueAtAsync(time.Start + TimeSpan.FromSeconds(1), Guard); + target.ForwardOpens.ShouldBe(1); + + time.Advance(TimeSpan.FromSeconds(1)); + await send.WaitAsync(Guard, TestCt); + target.ForwardOpens.ShouldBe(2); + } + + [Fact] + public async Task Concurrent_acquire_and_release_is_safe() + { + var pool = new DeviceSessionPool(new ManualTimeProvider()); + var keys = Enumerable.Range(0, 3).Select(i => new DeviceKey($"plc{i}", 44818, "1,0", "ControlLogix")).ToArray(); + var schedulers = new System.Collections.Concurrent.ConcurrentDictionary<(int, RequestScheduler), byte>(); + + var tasks = Enumerable.Range(0, 100).Select(i => Task.Run(async () => + { + var lease = pool.Acquire(keys[i % 3], Options()); + schedulers.TryAdd((i % 3, lease.Scheduler), 0); + await Task.Yield(); + await lease.DisposeAsync(); + }, TestCt)).ToArray(); + await Task.WhenAll(tasks).WaitAsync(Guard, TestCt); + + pool.EntryCount.ShouldBe(0); + // Every lease got a scheduler for its own key only, never another key's. + schedulers.Keys.GroupBy(k => k.Item2).ShouldAllBe(g => g.Select(k => k.Item1).Distinct().Count() == 1); + } + + [Fact] + public void Shared_is_one_process_wide_instance() => + DeviceSessionPool.Shared.ShouldBeSameAs(DeviceSessionPool.Shared); + + private sealed class CapturingLogger : ILogger + { + private int _warnings; + public int Warnings => Volatile.Read(ref _warnings); + public IDisposable? BeginScope(TState state) where TState : notnull => null; + public bool IsEnabled(LogLevel logLevel) => true; + + public void Log(LogLevel logLevel, EventId eventId, TState state, Exception? exception, + Func formatter) + { + if (logLevel == LogLevel.Warning) Interlocked.Increment(ref _warnings); + } + } +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipTarget.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipTarget.cs new file mode 100644 index 00000000..0ce9f4eb --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/FakeCipTarget.cs @@ -0,0 +1,123 @@ +using System.Buffers.Binary; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Cip; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Eip; +using ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Wire; +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes; + +/// +/// A CIP target on top of : answers Forward Open / Forward Close / +/// Unconnected Send / Direct (bare SendRRData) / connected SendUnitData requests and counts each kind. +/// +internal sealed class FakeCipTarget : IAsyncDisposable +{ + public const uint OTConnectionId = 0xC0FFEE01; + + private readonly List _forwardOpens = []; + private readonly List _forwardCloses = []; + private readonly List _unconnectedEmbedded = []; + private readonly List _direct = []; + private readonly List _connected = []; + + public FakeCipTarget() + { + Server.HandlerAsync = HandleAsync; + } + + public FakeEipServer Server { get; } = new(); + + public int Port => Server.Port; + + /// Forward Open request CIP → reply CIP. Defaults to . + public Func ForwardOpenReply { get; set; } = Accept; + + /// Reply to an ordinary service request (Direct, embedded in Unconnected Send, or connected). + /// Defaults to success with two data bytes (a Vendor ID, for the keep-alive). + public Func ServiceReply { get; set; } = req => [(byte)(req[0] | 0x80), 0, 0, 0, 0x01, 0x00]; + + /// Connected requests only: may delay, or return null to never answer. Defaults to . + public Func>? ConnectedHandler { get; set; } + + public int ForwardOpens { get { lock (_forwardOpens) return _forwardOpens.Count; } } + public int ForwardCloses { get { lock (_forwardCloses) return _forwardCloses.Count; } } + public int UnconnectedSends { get { lock (_unconnectedEmbedded) return _unconnectedEmbedded.Count; } } + public int DirectRequests { get { lock (_direct) return _direct.Count; } } + public int ConnectedRequests { get { lock (_connected) return _connected.Count; } } + + public IReadOnlyList ForwardOpenRequests { get { lock (_forwardOpens) return [.. _forwardOpens]; } } + public IReadOnlyList ForwardCloseRequests { get { lock (_forwardCloses) return [.. _forwardCloses]; } } + public IReadOnlyList UnconnectedEmbedded { get { lock (_unconnectedEmbedded) return [.. _unconnectedEmbedded]; } } + public IReadOnlyList DirectCips { get { lock (_direct) return [.. _direct]; } } + public IReadOnlyList ConnectedCips { get { lock (_connected) return [.. _connected]; } } + + /// A successful (Large) Forward Open reply echoing the request's ids and serials. + public static byte[] Accept(byte[] fo) => new CipWriter() + .U8((byte)(fo[0] | 0x80)).U8(0).U8(0).U8(0) + .U32(OTConnectionId) + .U32(BinaryPrimitives.ReadUInt32LittleEndian(fo.AsSpan(12))) + .U16(BinaryPrimitives.ReadUInt16LittleEndian(fo.AsSpan(16))) + .U16(BinaryPrimitives.ReadUInt16LittleEndian(fo.AsSpan(18))) + .U32(BinaryPrimitives.ReadUInt32LittleEndian(fo.AsSpan(20))) + .U32(2_000_000).U32(2_000_000).U8(0).U8(0) + .ToArray(); + + /// A refusal with the given general and extended status. + public static Func Refuse(byte general, params ushort[] extended) => fo => + { + var w = new CipWriter().U8((byte)(fo[0] | 0x80)).U8(0).U8(general).U8((byte)extended.Length); + foreach (var e in extended) w.U16(e); + return w.ToArray(); + }; + + public static bool IsForwardOpen(byte[] cip) => cip[0] is 0x54 or 0x5B && AddressesConnectionManager(cip); + + public static bool IsForwardClose(byte[] cip) => cip[0] == 0x4E && AddressesConnectionManager(cip); + + private static bool AddressesConnectionManager(byte[] cip) => + cip.Length >= 6 && cip.AsSpan(2, 4).SequenceEqual(CipPath.ConnectionManager); + + private async Task HandleAsync(byte[] frame) + { + var cip = FakeEipServer.RequestCip(frame); + if (FakeEipServer.Command(frame) == EncapCommand.SendUnitData) + { + lock (_connected) _connected.Add(cip); + var reply = ConnectedHandler is null ? ServiceReply(cip) : await ConnectedHandler(cip); + return reply is null ? null : FakeEipServer.UnitReply(frame, reply); + } + + if (IsForwardOpen(cip)) + { + lock (_forwardOpens) _forwardOpens.Add(cip); + return FakeEipServer.RRReply(frame, ForwardOpenReply(cip)); + } + if (IsForwardClose(cip)) + { + lock (_forwardCloses) _forwardCloses.Add(cip); + var w = new CipWriter().U8(0xCE).U8(0).U8(0).U8(0).Bytes(cip.AsSpan(8, 8)).U8(0).U8(0); + return FakeEipServer.RRReply(frame, w.ToArray()); + } + if (cip[0] == 0x52 && AddressesConnectionManager(cip)) + { + var len = BinaryPrimitives.ReadUInt16LittleEndian(cip.AsSpan(8)); + var embedded = cip.AsSpan(10, len).ToArray(); + lock (_unconnectedEmbedded) _unconnectedEmbedded.Add(embedded); + return FakeEipServer.RRReply(frame, ServiceReply(embedded)); + } + + lock (_direct) _direct.Add(cip); + return FakeEipServer.RRReply(frame, ServiceReply(cip)); + } + + /// Polls (real time) until holds. + public static async Task Until(Func condition, TimeSpan timeout, string what) + { + var giveUp = DateTime.UtcNow + timeout; + while (!condition()) + { + if (DateTime.UtcNow > giveUp) throw new TimeoutException($"Timed out waiting for {what}."); + await Task.Delay(5); + } + } + + public ValueTask DisposeAsync() => Server.DisposeAsync(); +} diff --git a/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/ManualTimeProvider.cs b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/ManualTimeProvider.cs new file mode 100644 index 00000000..49b5ee77 --- /dev/null +++ b/tests/Drivers/ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests/Fakes/ManualTimeProvider.cs @@ -0,0 +1,102 @@ +namespace ZB.MOM.WW.OtOpcUa.Driver.Ab.Transport.Tests.Fakes; + +/// +/// Deterministic : time moves only on , which fires every +/// timer falling due on the way, in due order, synchronously on the calling thread. +/// (Microsoft.Extensions.TimeProvider.Testing is not pinned in this repo and no package may be added.) +/// +internal sealed class ManualTimeProvider : TimeProvider +{ + private readonly object _lock = new(); + private readonly List _timers = []; + private DateTimeOffset _now; + + public ManualTimeProvider() => Start = _now = new DateTimeOffset(2026, 10, 7, 0, 0, 0, TimeSpan.Zero); + + public DateTimeOffset Start { get; } + + public override DateTimeOffset GetUtcNow() { lock (_lock) return _now; } + + public override long GetTimestamp() => GetUtcNow().UtcTicks; + + public override long TimestampFrequency => TimeSpan.TicksPerSecond; + + public override ITimer CreateTimer(TimerCallback callback, object? state, TimeSpan dueTime, TimeSpan period) + { + var t = new ManualTimer(this, callback, state); + lock (_lock) _timers.Add(t); + t.Change(dueTime, period); + return t; + } + + /// Armed timers (not disposed, with a due time). + public int ArmedTimers { get { lock (_lock) return _timers.Count(t => t.Due is not null); } } + + public bool HasTimerDueAt(DateTimeOffset at) { lock (_lock) return _timers.Any(t => t.Due == at); } + + /// Waits (in real time) until some code under test has armed a timer due at , + /// i.e. is genuinely parked on the fake clock rather than merely not yet scheduled. + public async Task WaitForTimerDueAtAsync(DateTimeOffset at, TimeSpan timeout) + { + var giveUp = DateTime.UtcNow + timeout; + while (!HasTimerDueAt(at)) + { + if (DateTime.UtcNow > giveUp) throw new TimeoutException($"No timer armed for {at:O} (now {GetUtcNow():O})."); + await Task.Delay(5); + } + } + + public void Advance(TimeSpan by) + { + DateTimeOffset target; + lock (_lock) target = _now + by; + while (true) + { + ManualTimer? next; + lock (_lock) + { + next = _timers.Where(t => t.Due <= target).MinBy(t => t.Due); + if (next is null) { _now = target; return; } + _now = next.Due!.Value; + next.Due = next.Period is { } p ? _now + p : null; + } + next.Fire(); + } + } + + private sealed class ManualTimer(ManualTimeProvider owner, TimerCallback callback, object? state) : ITimer + { + public DateTimeOffset? Due; // guarded by owner._lock + public TimeSpan? Period; // guarded by owner._lock + private bool _disposed; + + public void Fire() => callback(state); + + public bool Change(TimeSpan dueTime, TimeSpan period) + { + lock (owner._lock) + { + if (_disposed) return false; + Due = dueTime == Timeout.InfiniteTimeSpan ? null : owner._now + dueTime; + Period = period == Timeout.InfiniteTimeSpan || period == TimeSpan.Zero ? null : period; + return true; + } + } + + public void Dispose() + { + lock (owner._lock) + { + _disposed = true; + Due = null; + owner._timers.Remove(this); + } + } + + public ValueTask DisposeAsync() + { + Dispose(); + return ValueTask.CompletedTask; + } + } +}