diff --git a/src/AR.Iec61850.Transports.Npcap/AR.Iec61850.Transports.Npcap.csproj b/src/AR.Iec61850.Transports.Npcap/AR.Iec61850.Transports.Npcap.csproj index 680abfa..0162033 100644 --- a/src/AR.Iec61850.Transports.Npcap/AR.Iec61850.Transports.Npcap.csproj +++ b/src/AR.Iec61850.Transports.Npcap/AR.Iec61850.Transports.Npcap.csproj @@ -8,6 +8,10 @@ + + + + Copyright (C) 2026 Mas Ari / masarray GPL-3.0-or-later diff --git a/src/AR.Iec61850.Transports.Npcap/NpcapProcessBusDuplexTransport.cs b/src/AR.Iec61850.Transports.Npcap/NpcapProcessBusDuplexTransport.cs index cd2c669..07855ac 100644 --- a/src/AR.Iec61850.Transports.Npcap/NpcapProcessBusDuplexTransport.cs +++ b/src/AR.Iec61850.Transports.Npcap/NpcapProcessBusDuplexTransport.cs @@ -20,8 +20,9 @@ public sealed class NpcapProcessBusDuplexTransport : IProcessBusTransport, IProc private readonly ICaptureDevice _device; private readonly IInjectionDevice _injectionDevice; - private readonly object _gate = new(); - private readonly SemaphoreSlim _sendGate = new(1, 1); + private readonly object _captureGate = new(); + private readonly object _clockMapGate = new(); + private readonly SemaphoreSlim _injectionGate = new(1, 1); private readonly Dictionary _svTransmitClocks = new(); private bool _capturing; private bool _disposed; @@ -45,21 +46,23 @@ public async ValueTask SendAsync(ReadOnlyMemory frame, CancellationToken c ObjectDisposedException.ThrowIf(_disposed, this); cancellationToken.ThrowIfCancellationRequested(); - await _sendGate.WaitAsync(cancellationToken).ConfigureAwait(false); - try + if (!TryReadSampledValuesKey(frame.Span, out var streamKey)) { - var isSampledValues = TryReadSampledValuesKey(frame.Span, out var streamKey); - if (isSampledValues) - await PaceSampledValuesAsync(streamKey, cancellationToken).ConfigureAwait(false); - - _injectionDevice.SendPacket(frame.ToArray()); + await InjectAsync(frame, cancellationToken).ConfigureAwait(false); + return; + } - if (isSampledValues) - CommitSampledValuesSend(streamKey, Stopwatch.GetTimestamp()); + var clock = GetOrCreateClock(streamKey); + await clock.PacingGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await PaceSampledValuesAsync(clock, cancellationToken).ConfigureAwait(false); + await InjectAsync(frame, cancellationToken).ConfigureAwait(false); + clock.Commit(Stopwatch.GetTimestamp()); } finally { - _sendGate.Release(); + clock.PacingGate.Release(); } } @@ -83,7 +86,7 @@ public async IAsyncEnumerable CaptureAsync( try { - lock (_gate) + lock (_captureGate) { if (_capturing) throw new InvalidOperationException("This Npcap session is already capturing."); @@ -130,7 +133,7 @@ public async IAsyncEnumerable CaptureAsync( } } - lock (_gate) + lock (_captureGate) _capturing = false; channel.Writer.TryComplete(); @@ -151,16 +154,56 @@ public void Dispose() // Best-effort cleanup only. } - _sendGate.Dispose(); + _injectionGate.Dispose(); + lock (_clockMapGate) + { + foreach (var clock in _svTransmitClocks.Values) + clock.Dispose(); + _svTransmitClocks.Clear(); + } + _disposed = true; } - private async ValueTask PaceSampledValuesAsync(SvTransmitKey key, CancellationToken cancellationToken) + private async ValueTask InjectAsync(ReadOnlyMemory frame, CancellationToken cancellationToken) + { + // Keep the device critical section intentionally short. PTP and GOOSE may pass + // while another SV stream is waiting for its pacing deadline. + await _injectionGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + _injectionDevice.SendPacket(frame.ToArray()); + } + finally + { + _injectionGate.Release(); + } + } + + private SvTransmitClock GetOrCreateClock(SvTransmitKey key) { - if (!_svTransmitClocks.TryGetValue(key, out var clock) || clock.NominalIntervalTicks <= 0) + lock (_clockMapGate) + { + if (_svTransmitClocks.TryGetValue(key, out var existing)) + return existing; + + var created = new SvTransmitClock( + MinimumLearnableIntervalTicks, + MaximumLearnableIntervalTicks); + _svTransmitClocks.Add(key, created); + return created; + } + } + + private static async ValueTask PaceSampledValuesAsync( + SvTransmitClock clock, + CancellationToken cancellationToken) + { + var intervalTicks = clock.NominalIntervalTicks; + if (clock.LastSentTicks <= 0 || intervalTicks <= 0) return; - var targetTicks = clock.LastSentTicks + clock.NominalIntervalTicks; + var targetTicks = clock.LastSentTicks + intervalTicks; while (true) { cancellationToken.ThrowIfCancellationRequested(); @@ -183,30 +226,6 @@ await Task.Delay( } } - private void CommitSampledValuesSend(SvTransmitKey key, long sentTicks) - { - if (!_svTransmitClocks.TryGetValue(key, out var clock)) - { - _svTransmitClocks[key] = new SvTransmitClock(sentTicks, 0); - return; - } - - var observedInterval = sentTicks - clock.LastSentTicks; - var nominalInterval = clock.NominalIntervalTicks; - var learnable = observedInterval >= MinimumLearnableIntervalTicks && - observedInterval <= MaximumLearnableIntervalTicks && - (nominalInterval <= 0 || observedInterval <= nominalInterval * 3); - - if (learnable) - { - nominalInterval = nominalInterval <= 0 - ? observedInterval - : (long)Math.Round((nominalInterval * 0.9) + (observedInterval * 0.1)); - } - - _svTransmitClocks[key] = new SvTransmitClock(sentTicks, nominalInterval); - } - private static bool TryReadSampledValuesKey(ReadOnlySpan frame, out SvTransmitKey key) { key = default; @@ -245,5 +264,27 @@ private static DateTimeOffset ToDateTimeOffset(PosixTimeval timeval) } private readonly record struct SvTransmitKey(ulong MacPrefix, uint MacSuffix, ushort AppId, ushort VlanId); - private sealed record SvTransmitClock(long LastSentTicks, long NominalIntervalTicks); + + private sealed class SvTransmitClock : IDisposable + { + private readonly SvTransmitIntervalEstimator _estimator; + + public SvTransmitClock(long minimumIntervalTicks, long maximumIntervalTicks) + { + _estimator = new SvTransmitIntervalEstimator(minimumIntervalTicks, maximumIntervalTicks); + } + + public SemaphoreSlim PacingGate { get; } = new(1, 1); + public long LastSentTicks { get; private set; } + public long NominalIntervalTicks => _estimator.NominalIntervalTicks; + + public void Commit(long sentTicks) + { + if (LastSentTicks > 0) + _estimator.Observe(sentTicks - LastSentTicks); + LastSentTicks = sentTicks; + } + + public void Dispose() => PacingGate.Dispose(); + } } diff --git a/src/AR.Iec61850.Transports.Npcap/SvTransmitIntervalEstimator.cs b/src/AR.Iec61850.Transports.Npcap/SvTransmitIntervalEstimator.cs new file mode 100644 index 0000000..8394687 --- /dev/null +++ b/src/AR.Iec61850.Transports.Npcap/SvTransmitIntervalEstimator.cs @@ -0,0 +1,89 @@ +namespace AR.Iec61850.Transports.Npcap; + +/// +/// Learns a stable SV transmit interval without allowing one scheduler-late observation +/// to become the permanent wire rate. The estimator activates only after several +/// mutually consistent intervals have been observed. +/// +internal sealed class SvTransmitIntervalEstimator +{ + private readonly long _minimumIntervalTicks; + private readonly long _maximumIntervalTicks; + private readonly int _requiredConsistentIntervals; + private long _candidateIntervalTicks; + private int _candidateCount; + + public SvTransmitIntervalEstimator( + long minimumIntervalTicks, + long maximumIntervalTicks, + int requiredConsistentIntervals = 4) + { + if (minimumIntervalTicks <= 0) + throw new ArgumentOutOfRangeException(nameof(minimumIntervalTicks)); + if (maximumIntervalTicks < minimumIntervalTicks) + throw new ArgumentOutOfRangeException(nameof(maximumIntervalTicks)); + if (requiredConsistentIntervals < 2) + throw new ArgumentOutOfRangeException(nameof(requiredConsistentIntervals)); + + _minimumIntervalTicks = minimumIntervalTicks; + _maximumIntervalTicks = maximumIntervalTicks; + _requiredConsistentIntervals = requiredConsistentIntervals; + } + + public long NominalIntervalTicks { get; private set; } + public int CandidateCount => _candidateCount; + + public void Observe(long intervalTicks) + { + if (intervalTicks < _minimumIntervalTicks || intervalTicks > _maximumIntervalTicks) + { + if (NominalIntervalTicks == 0) + ResetCandidate(); + return; + } + + if (NominalIntervalTicks > 0) + { + var minimumAccepted = NominalIntervalTicks * 3 / 4; + var maximumAccepted = NominalIntervalTicks * 3 / 2; + if (intervalTicks >= minimumAccepted && intervalTicks <= maximumAccepted) + { + NominalIntervalTicks = (long)Math.Round( + (NominalIntervalTicks * 0.9) + (intervalTicks * 0.1)); + } + + return; + } + + if (_candidateCount == 0) + { + _candidateIntervalTicks = intervalTicks; + _candidateCount = 1; + return; + } + + var tolerance = Math.Max( + _minimumIntervalTicks / 2, + (long)Math.Round(_candidateIntervalTicks * 0.15)); + if (Math.Abs(intervalTicks - _candidateIntervalTicks) > tolerance) + { + _candidateIntervalTicks = intervalTicks; + _candidateCount = 1; + return; + } + + _candidateIntervalTicks = (long)Math.Round( + ((_candidateIntervalTicks * _candidateCount) + intervalTicks) / + (double)(_candidateCount + 1)); + _candidateCount++; + + if (_candidateCount >= _requiredConsistentIntervals) + NominalIntervalTicks = _candidateIntervalTicks; + } + + private void ResetCandidate() + { + _candidateIntervalTicks = 0; + _candidateCount = 0; + } +} diff --git a/tests/AR.Iec61850.Tests/AR.Iec61850.Tests.csproj b/tests/AR.Iec61850.Tests/AR.Iec61850.Tests.csproj index 8942a5d..08232fc 100644 --- a/tests/AR.Iec61850.Tests/AR.Iec61850.Tests.csproj +++ b/tests/AR.Iec61850.Tests/AR.Iec61850.Tests.csproj @@ -25,6 +25,7 @@ + diff --git a/tests/AR.Iec61850.Tests/Transports/SvTransmitIntervalEstimatorTests.cs b/tests/AR.Iec61850.Tests/Transports/SvTransmitIntervalEstimatorTests.cs new file mode 100644 index 0000000..75b88da --- /dev/null +++ b/tests/AR.Iec61850.Tests/Transports/SvTransmitIntervalEstimatorTests.cs @@ -0,0 +1,61 @@ +using AR.Iec61850.Transports.Npcap; + +namespace AR.Iec61850.Tests.Transports; + +public sealed class SvTransmitIntervalEstimatorTests +{ + [Fact] + public void SingleLateObservation_DoesNotBecomePermanentNominalRate() + { + var estimator = new SvTransmitIntervalEstimator(20, 5_000, requiredConsistentIntervals: 4); + + estimator.Observe(1_000); // Scheduler-late first interval. + Assert.Equal(0, estimator.NominalIntervalTicks); + + estimator.Observe(250); + estimator.Observe(248); + estimator.Observe(252); + Assert.Equal(0, estimator.NominalIntervalTicks); + + estimator.Observe(251); + + Assert.InRange(estimator.NominalIntervalTicks, 248, 252); + } + + [Fact] + public void InconsistentIntervals_DoNotActivatePacing() + { + var estimator = new SvTransmitIntervalEstimator(20, 5_000, requiredConsistentIntervals: 4); + + foreach (var interval in new long[] { 250, 800, 240, 1_200, 260, 700 }) + estimator.Observe(interval); + + Assert.Equal(0, estimator.NominalIntervalTicks); + Assert.Equal(1, estimator.CandidateCount); + } + + [Fact] + public void ActiveNominalRate_IgnoresLongSchedulerStall() + { + var estimator = new SvTransmitIntervalEstimator(20, 5_000, requiredConsistentIntervals: 4); + foreach (var interval in new long[] { 250, 249, 251, 250 }) + estimator.Observe(interval); + + var nominalBeforeStall = estimator.NominalIntervalTicks; + estimator.Observe(4_000); + + Assert.Equal(nominalBeforeStall, estimator.NominalIntervalTicks); + } + + [Fact] + public void ActiveNominalRate_TracksOnlyNearbySteadyIntervals() + { + var estimator = new SvTransmitIntervalEstimator(20, 5_000, requiredConsistentIntervals: 4); + foreach (var interval in new long[] { 250, 250, 250, 250 }) + estimator.Observe(interval); + + estimator.Observe(260); + + Assert.InRange(estimator.NominalIntervalTicks, 250, 252); + } +}