From 04fa0a629614ee996e648a4c2d6ba27bf6a96766 Mon Sep 17 00:00:00 2001 From: masarray Date: Mon, 3 Aug 2026 13:52:17 +0700 Subject: [PATCH 1/2] fix: pace late SV transmission instead of catch-up bursting --- .../NpcapProcessBusDuplexTransport.cs | 97 +++++++++++++++++-- 1 file changed, 90 insertions(+), 7 deletions(-) diff --git a/src/AR.Iec61850.Transports.Npcap/NpcapProcessBusDuplexTransport.cs b/src/AR.Iec61850.Transports.Npcap/NpcapProcessBusDuplexTransport.cs index 6ee800f..7b33dd0 100644 --- a/src/AR.Iec61850.Transports.Npcap/NpcapProcessBusDuplexTransport.cs +++ b/src/AR.Iec61850.Transports.Npcap/NpcapProcessBusDuplexTransport.cs @@ -1,5 +1,7 @@ +using System.Diagnostics; using System.Runtime.CompilerServices; using System.Threading.Channels; +using AR.Iec61850.SampledValues; using AR.Iec61850.Transports; using SharpPcap; @@ -14,6 +16,8 @@ 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 Dictionary _svTransmitClocks = new(StringComparer.Ordinal); private bool _capturing; private bool _disposed; @@ -31,12 +35,22 @@ public NpcapProcessBusDuplexTransport(ICaptureDevice device) _device.Open(DeviceModes.Promiscuous, 1000); } - public ValueTask SendAsync(ReadOnlyMemory frame, CancellationToken cancellationToken = default) + public async ValueTask SendAsync(ReadOnlyMemory frame, CancellationToken cancellationToken = default) { ObjectDisposedException.ThrowIf(_disposed, this); cancellationToken.ThrowIfCancellationRequested(); - _injectionDevice.SendPacket(frame.ToArray()); - return ValueTask.CompletedTask; + + await _sendGate.WaitAsync(cancellationToken).ConfigureAwait(false); + try + { + await PaceSampledValuesAsync(frame, cancellationToken).ConfigureAwait(false); + _injectionDevice.SendPacket(frame.ToArray()); + CommitSampledValuesSend(frame, Stopwatch.GetTimestamp()); + } + finally + { + _sendGate.Release(); + } } public async IAsyncEnumerable CaptureAsync( @@ -72,22 +86,22 @@ public async IAsyncEnumerable CaptureAsync( handler = (_, capture) => { - var frame = new ProcessBusCapturedFrame + var capturedFrame = new ProcessBusCapturedFrame { Timestamp = ToDateTimeOffset(capture.Header.Timeval), Frame = capture.Data.ToArray(), Source = _device.Name ?? string.Empty }; - channel.Writer.TryWrite(frame); + channel.Writer.TryWrite(capturedFrame); }; _device.OnPacketArrival += handler; _device.StartCapture(); started = true; - await foreach (var frame in channel.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) - yield return frame; + await foreach (var capturedFrame in channel.Reader.ReadAllAsync(cancellationToken).ConfigureAwait(false)) + yield return capturedFrame; } finally { @@ -127,13 +141,82 @@ public void Dispose() // Best-effort cleanup only. } + _sendGate.Dispose(); _disposed = true; } + private async ValueTask PaceSampledValuesAsync(ReadOnlyMemory frame, CancellationToken cancellationToken) + { + if (!TryGetSampledValuesClock(frame, out var key, out var referenceTime)) + return; + + if (!_svTransmitClocks.TryGetValue(key, out var clock)) + return; + + var referenceInterval = referenceTime - clock.ReferenceTime; + if (referenceInterval <= TimeSpan.Zero || referenceInterval > TimeSpan.FromMilliseconds(100)) + return; + + var intervalTicks = (long)Math.Round(referenceInterval.TotalSeconds * Stopwatch.Frequency); + if (intervalTicks <= 0) + return; + + var targetTicks = clock.SentTicks + intervalTicks; + while (true) + { + cancellationToken.ThrowIfCancellationRequested(); + var remainingTicks = targetTicks - Stopwatch.GetTimestamp(); + if (remainingTicks <= 0) + return; + + var remainingMilliseconds = remainingTicks * 1000.0 / Stopwatch.Frequency; + if (remainingMilliseconds > 2) + { + await Task.Delay( + TimeSpan.FromMilliseconds(Math.Min(remainingMilliseconds - 1, 10)), + cancellationToken) + .ConfigureAwait(false); + } + else + { + Thread.SpinWait(64); + } + } + } + + private void CommitSampledValuesSend(ReadOnlyMemory frame, long sentTicks) + { + if (!TryGetSampledValuesClock(frame, out var key, out var referenceTime)) + return; + + _svTransmitClocks[key] = new SvTransmitClock(referenceTime, sentTicks); + } + + private static bool TryGetSampledValuesClock( + ReadOnlyMemory frameBytes, + out string key, + out DateTimeOffset referenceTime) + { + key = string.Empty; + referenceTime = default; + + if (!SampledValuesFrameParser.TryParseEthernetFrame(frameBytes, out var frame) || + frame.Pdu.Asdus.FirstOrDefault() is not { ReferenceTime: { } time } first) + { + return false; + } + + referenceTime = time.Value; + key = $"{frame.Source}|{frame.Destination}|{frame.Vlan?.VlanId.ToString() ?? "-"}|{frame.AppId:X4}|{first.SvId}"; + return true; + } + private static DateTimeOffset ToDateTimeOffset(PosixTimeval timeval) { var seconds = Convert.ToInt64(timeval.Seconds); var microseconds = Convert.ToInt64(timeval.MicroSeconds); return DateTimeOffset.FromUnixTimeSeconds(seconds).AddTicks(checked(microseconds * 10)); } + + private sealed record SvTransmitClock(DateTimeOffset ReferenceTime, long SentTicks); } From 674d351017457d782f1f77638483d4b4701a7071 Mon Sep 17 00:00:00 2001 From: masarray Date: Mon, 3 Aug 2026 13:54:58 +0700 Subject: [PATCH 2/2] perf: use allocation-free SV stream pacing key --- .../NpcapProcessBusDuplexTransport.cs | 93 ++++++++++++------- 1 file changed, 60 insertions(+), 33 deletions(-) diff --git a/src/AR.Iec61850.Transports.Npcap/NpcapProcessBusDuplexTransport.cs b/src/AR.Iec61850.Transports.Npcap/NpcapProcessBusDuplexTransport.cs index 7b33dd0..cd2c669 100644 --- a/src/AR.Iec61850.Transports.Npcap/NpcapProcessBusDuplexTransport.cs +++ b/src/AR.Iec61850.Transports.Npcap/NpcapProcessBusDuplexTransport.cs @@ -1,7 +1,7 @@ +using System.Buffers.Binary; using System.Diagnostics; using System.Runtime.CompilerServices; using System.Threading.Channels; -using AR.Iec61850.SampledValues; using AR.Iec61850.Transports; using SharpPcap; @@ -13,11 +13,16 @@ namespace AR.Iec61850.Transports.Npcap; /// public sealed class NpcapProcessBusDuplexTransport : IProcessBusTransport, IProcessBusFrameSource, IDisposable { + private const ushort VlanEtherType = 0x8100; + private const ushort SampledValuesEtherType = 0x88BA; + private static readonly long MinimumLearnableIntervalTicks = Math.Max(1, Stopwatch.Frequency / 50_000); // 20 us + private static readonly long MaximumLearnableIntervalTicks = Math.Max(1, Stopwatch.Frequency / 200); // 5 ms + private readonly ICaptureDevice _device; private readonly IInjectionDevice _injectionDevice; private readonly object _gate = new(); private readonly SemaphoreSlim _sendGate = new(1, 1); - private readonly Dictionary _svTransmitClocks = new(StringComparer.Ordinal); + private readonly Dictionary _svTransmitClocks = new(); private bool _capturing; private bool _disposed; @@ -43,9 +48,14 @@ public async ValueTask SendAsync(ReadOnlyMemory frame, CancellationToken c await _sendGate.WaitAsync(cancellationToken).ConfigureAwait(false); try { - await PaceSampledValuesAsync(frame, cancellationToken).ConfigureAwait(false); + var isSampledValues = TryReadSampledValuesKey(frame.Span, out var streamKey); + if (isSampledValues) + await PaceSampledValuesAsync(streamKey, cancellationToken).ConfigureAwait(false); + _injectionDevice.SendPacket(frame.ToArray()); - CommitSampledValuesSend(frame, Stopwatch.GetTimestamp()); + + if (isSampledValues) + CommitSampledValuesSend(streamKey, Stopwatch.GetTimestamp()); } finally { @@ -145,23 +155,12 @@ public void Dispose() _disposed = true; } - private async ValueTask PaceSampledValuesAsync(ReadOnlyMemory frame, CancellationToken cancellationToken) + private async ValueTask PaceSampledValuesAsync(SvTransmitKey key, CancellationToken cancellationToken) { - if (!TryGetSampledValuesClock(frame, out var key, out var referenceTime)) - return; - - if (!_svTransmitClocks.TryGetValue(key, out var clock)) + if (!_svTransmitClocks.TryGetValue(key, out var clock) || clock.NominalIntervalTicks <= 0) return; - var referenceInterval = referenceTime - clock.ReferenceTime; - if (referenceInterval <= TimeSpan.Zero || referenceInterval > TimeSpan.FromMilliseconds(100)) - return; - - var intervalTicks = (long)Math.Round(referenceInterval.TotalSeconds * Stopwatch.Frequency); - if (intervalTicks <= 0) - return; - - var targetTicks = clock.SentTicks + intervalTicks; + var targetTicks = clock.LastSentTicks + clock.NominalIntervalTicks; while (true) { cancellationToken.ThrowIfCancellationRequested(); @@ -184,30 +183,57 @@ await Task.Delay( } } - private void CommitSampledValuesSend(ReadOnlyMemory frame, long sentTicks) + private void CommitSampledValuesSend(SvTransmitKey key, long sentTicks) { - if (!TryGetSampledValuesClock(frame, out var key, out var referenceTime)) + 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(referenceTime, sentTicks); + _svTransmitClocks[key] = new SvTransmitClock(sentTicks, nominalInterval); } - private static bool TryGetSampledValuesClock( - ReadOnlyMemory frameBytes, - out string key, - out DateTimeOffset referenceTime) + private static bool TryReadSampledValuesKey(ReadOnlySpan frame, out SvTransmitKey key) { - key = string.Empty; - referenceTime = default; + key = default; + if (frame.Length < 22) + return false; - if (!SampledValuesFrameParser.TryParseEthernetFrame(frameBytes, out var frame) || - frame.Pdu.Asdus.FirstOrDefault() is not { ReferenceTime: { } time } first) + var etherType = BinaryPrimitives.ReadUInt16BigEndian(frame.Slice(12, 2)); + var processBusOffset = 14; + ushort vlanId = 0; + if (etherType == VlanEtherType) { - return false; + if (frame.Length < 26) + return false; + + vlanId = (ushort)(BinaryPrimitives.ReadUInt16BigEndian(frame.Slice(14, 2)) & 0x0FFF); + etherType = BinaryPrimitives.ReadUInt16BigEndian(frame.Slice(16, 2)); + processBusOffset = 18; } - referenceTime = time.Value; - key = $"{frame.Source}|{frame.Destination}|{frame.Vlan?.VlanId.ToString() ?? "-"}|{frame.AppId:X4}|{first.SvId}"; + if (etherType != SampledValuesEtherType || frame.Length < processBusOffset + 2) + return false; + + key = new SvTransmitKey( + BinaryPrimitives.ReadUInt64BigEndian(frame.Slice(0, 8)), + BinaryPrimitives.ReadUInt32BigEndian(frame.Slice(8, 4)), + BinaryPrimitives.ReadUInt16BigEndian(frame.Slice(processBusOffset, 2)), + vlanId); return true; } @@ -218,5 +244,6 @@ private static DateTimeOffset ToDateTimeOffset(PosixTimeval timeval) return DateTimeOffset.FromUnixTimeSeconds(seconds).AddTicks(checked(microseconds * 10)); } - private sealed record SvTransmitClock(DateTimeOffset ReferenceTime, long SentTicks); + private readonly record struct SvTransmitKey(ulong MacPrefix, uint MacSuffix, ushort AppId, ushort VlanId); + private sealed record SvTransmitClock(long LastSentTicks, long NominalIntervalTicks); }