feat(ab-transport): per-device connection with routing modes, shared scheduler, keep-alive and Forward Open throttle

2026-10-07 16:35:11 -04:00
parent d7145ddfa5
commit b82fffd269
11 changed files with 1535 additions and 0 deletions
@@ -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;
+
+/// <summary>
+/// Process-wide owner of device connections: exactly one <see cref="CipConnection"/> and one
+/// <see cref="RequestScheduler"/> per <see cref="DeviceKey"/> (review B1), handed out as ref-counted
+/// <see cref="ConnectionLease"/>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.
+/// </summary>
+/// <remarks>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.</remarks>
+public sealed class DeviceSessionPool
+{
+    /// <summary>The process-wide pool the drivers use. Tests build their own.</summary>
+    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<DeviceKey, Entry> _entries = new();
+    private readonly Dictionary<DeviceKey, ForwardOpenThrottle> _throttles = new();
+    private readonly Dictionary<DeviceKey, Task> _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;
+    }
+
+    /// <summary>Test hook: keys with a live entry.</summary>
+    internal int EntryCount { get { lock (_lock) return _entries.Count; } }
+
+    /// <summary>Lease the shared connection for <paramref name="key"/>, 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 <paramref name="options"/> is logged once per entry.</summary>
+    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
+        }
+    }
+}
+
+/// <summary>A reference on a device's shared connection. Dispose exactly once (further disposes are no-ops).</summary>
+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;
+
+    /// <summary>The device's channel. Prefer <see cref="Scheduler"/>: it serializes and packs requests.</summary>
+    public ICipChannel Channel => _entry.Connection;
+
+    /// <summary>The device's single request scheduler, shared by every lease on the key.</summary>
+    public RequestScheduler Scheduler => _entry.Scheduler;
+
+    public ValueTask DisposeAsync() =>
+        Interlocked.Exchange(ref _disposed, 1) == 0 ? _pool.ReleaseAsync(_key, _entry) : ValueTask.CompletedTask;
+}