From eeba7de76bfb6eea559c23d32049aec028d105be Mon Sep 17 00:00:00 2001 From: Ari Sulistiono Date: Wed, 7 Oct 2026 16:18:56 +0700 Subject: [PATCH 1/4] perf(reporting): add event-driven InformationReport queue wake (#144) --- src/AR.Iec61850/Mms/MmsReceiveRouter.cs | 43 +++++++++++++++++++++++++ 1 file changed, 43 insertions(+) diff --git a/src/AR.Iec61850/Mms/MmsReceiveRouter.cs b/src/AR.Iec61850/Mms/MmsReceiveRouter.cs index 052909bc..402ad4fc 100644 --- a/src/AR.Iec61850/Mms/MmsReceiveRouter.cs +++ b/src/AR.Iec61850/Mms/MmsReceiveRouter.cs @@ -25,6 +25,7 @@ public sealed class MmsReceiveRouter private readonly Queue _unconfirmed = new(); private readonly Queue _unmatched = new(); private readonly Dictionary> _informationReportSubscriptions = new(); + private readonly SemaphoreSlim _informationReportAvailable = new(0, 1); public int QueuedConfirmedResultCount { @@ -88,6 +89,7 @@ public MmsReceiveRouteResult Route(ReadOnlyMemory presentationPayload) if (envelope.IsInformationReport) { _informationReports.Enqueue(envelope); + SignalInformationReportAvailable(); foreach (var subscription in _informationReportSubscriptions.Values) subscription.Writer.TryWrite(envelope); @@ -152,6 +154,31 @@ public bool TryDequeueInformationReport(out MmsPduEnvelope envelope) return false; } + /// + /// Waits until the canonical InformationReport queue has work without polling. + /// The signal is binary: multiple reports coalesce into one wake because callers + /// drain the queue after waking. Queue contents remain the single routing authority. + /// + internal async Task WaitForInformationReportAsync( + TimeSpan timeout, + CancellationToken cancellationToken = default) + { + cancellationToken.ThrowIfCancellationRequested(); + + if (QueuedInformationReportCount > 0) + { + _informationReportAvailable.Wait(0); + return true; + } + + if (timeout <= TimeSpan.Zero) + return false; + + return await _informationReportAvailable + .WaitAsync(timeout, cancellationToken) + .ConfigureAwait(false); + } + internal MmsInformationReportSubscription SubscribeInformationReports(int capacity = 32) { var boundedCapacity = Math.Clamp(capacity, 1, 1024); @@ -197,6 +224,18 @@ internal void FaultInformationReportSubscriptions(Exception exception) subscription.Writer.TryComplete(exception); } + private void SignalInformationReportAvailable() + { + try + { + _informationReportAvailable.Release(); + } + catch (SemaphoreFullException) + { + // A wake is already pending. The queue remains authoritative and will be drained. + } + } + public void Clear() { Channel[] subscriptions; @@ -210,6 +249,10 @@ public void Clear() _informationReportSubscriptions.Clear(); } + while (_informationReportAvailable.Wait(0)) + { + } + foreach (var subscription in subscriptions) subscription.Writer.TryComplete(new IOException("MMS association receive routing was reset.")); } From 53e7e4b2f993b82f5941b6afb5642db540fa2c76 Mon Sep 17 00:00:00 2001 From: Ari Sulistiono Date: Wed, 7 Oct 2026 16:18:59 +0700 Subject: [PATCH 2/4] perf(reporting): wake persistent monitors from report arrival instead of 25ms polling (#144) --- .../Mms/MmsPersistentReportMonitor.cs | 23 ++++++++++++++++--- 1 file changed, 20 insertions(+), 3 deletions(-) diff --git a/src/AR.Iec61850/Mms/MmsPersistentReportMonitor.cs b/src/AR.Iec61850/Mms/MmsPersistentReportMonitor.cs index 5a44096e..403ed96f 100644 --- a/src/AR.Iec61850/Mms/MmsPersistentReportMonitor.cs +++ b/src/AR.Iec61850/Mms/MmsPersistentReportMonitor.cs @@ -575,9 +575,26 @@ private async Task ReceiveAndDispatchPersistentReportsAsync( if (IsReceivePumpRunning) { - var delay = remaining < TimeSpan.FromMilliseconds(25) ? remaining : TimeSpan.FromMilliseconds(25); - if (delay > TimeSpan.Zero) - await Task.Delay(delay, cancellationToken).ConfigureAwait(false); + // The association-owned receive pump already blocks on the transport. + // Do not add a 25 ms application polling loop on top of it. Wait for the + // router's binary InformationReport signal, bounded by the next legitimate + // scheduler deadline (residual polling) or this receive slice deadline. + var wakeAt = deadline; + if (pollDirectory != null && + references.Length > 0 && + nextPollAt < wakeAt) + { + wakeAt = nextPollAt; + } + + var waitFor = wakeAt - DateTimeOffset.UtcNow; + if (waitFor > TimeSpan.Zero) + { + await _receiveRouter.WaitForInformationReportAsync( + waitFor, + cancellationToken).ConfigureAwait(false); + } + continue; } From fb78308d0dc80c9d8d8ba05ce3686688c612bd7a Mon Sep 17 00:00:00 2001 From: Ari Sulistiono Date: Wed, 7 Oct 2026 16:19:02 +0700 Subject: [PATCH 3/4] test(reporting): lock event-driven InformationReport wake semantics (#144) --- .../Mms/MmsReceiveRouterTests.cs | 48 +++++++++++++++++++ 1 file changed, 48 insertions(+) diff --git a/tests/AR.Iec61850.Tests/Mms/MmsReceiveRouterTests.cs b/tests/AR.Iec61850.Tests/Mms/MmsReceiveRouterTests.cs index 31451bd1..377e2144 100644 --- a/tests/AR.Iec61850.Tests/Mms/MmsReceiveRouterTests.cs +++ b/tests/AR.Iec61850.Tests/Mms/MmsReceiveRouterTests.cs @@ -99,6 +99,54 @@ public async Task Route_FansOutInformationReportWithoutStealingLegacyQueue() Assert.True(queued.IsInformationReport); } + [Fact] + public async Task WaitForInformationReportAsync_ReturnsImmediately_WhenQueueAlreadyHasReport() + { + var router = new MmsReceiveRouter(); + router.Route(BuildInformationReport()); + + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1)); + var available = await router.WaitForInformationReportAsync( + TimeSpan.FromSeconds(1), + timeout.Token); + + Assert.True(available); + Assert.True(router.TryDequeueInformationReport(out var queued)); + Assert.True(queued.IsInformationReport); + } + + [Fact] + public async Task WaitForInformationReportAsync_WakesFromFutureReport_WithoutPolling() + { + var router = new MmsReceiveRouter(); + + using var timeout = new CancellationTokenSource(TimeSpan.FromSeconds(1)); + var wait = router.WaitForInformationReportAsync( + TimeSpan.FromSeconds(1), + timeout.Token); + + await Task.Yield(); + Assert.False(wait.IsCompleted); + + router.Route(BuildInformationReport()); + + Assert.True(await wait); + Assert.True(router.TryDequeueInformationReport(out var queued)); + Assert.True(queued.IsInformationReport); + } + + [Fact] + public async Task WaitForInformationReportAsync_TimesOut_WhenNoReportArrives() + { + var router = new MmsReceiveRouter(); + + var available = await router.WaitForInformationReportAsync( + TimeSpan.FromMilliseconds(20)); + + Assert.False(available); + Assert.Equal(0, router.QueuedInformationReportCount); + } + [Fact] public async Task Clear_FaultsActiveInformationReportSubscribers() { From 397aa8254ab3eedb81a65702e0dc8db0a3f5186e Mon Sep 17 00:00:00 2001 From: Ari Sulistiono Date: Wed, 7 Oct 2026 16:20:27 +0700 Subject: [PATCH 4/4] test(reporting): prove burst reports coalesce without stale wake (#144) --- .../Mms/MmsReceiveRouterTests.cs | 15 +++++++++++++++ 1 file changed, 15 insertions(+) diff --git a/tests/AR.Iec61850.Tests/Mms/MmsReceiveRouterTests.cs b/tests/AR.Iec61850.Tests/Mms/MmsReceiveRouterTests.cs index 377e2144..4910d871 100644 --- a/tests/AR.Iec61850.Tests/Mms/MmsReceiveRouterTests.cs +++ b/tests/AR.Iec61850.Tests/Mms/MmsReceiveRouterTests.cs @@ -135,6 +135,21 @@ public async Task WaitForInformationReportAsync_WakesFromFutureReport_WithoutPol Assert.True(queued.IsInformationReport); } + [Fact] + public async Task WaitForInformationReportAsync_CoalescesBurstIntoSingleWake() + { + var router = new MmsReceiveRouter(); + router.Route(BuildInformationReport()); + router.Route(BuildInformationReport()); + + Assert.True(await router.WaitForInformationReportAsync(TimeSpan.FromSeconds(1))); + Assert.True(router.TryDequeueInformationReport(out _)); + Assert.True(router.TryDequeueInformationReport(out _)); + + var staleWake = await router.WaitForInformationReportAsync(TimeSpan.FromMilliseconds(20)); + Assert.False(staleWake); + } + [Fact] public async Task WaitForInformationReportAsync_TimesOut_WhenNoReportArrives() {