From 27b6ad0d0a48b7b6bde79847bca2218c6b4155de Mon Sep 17 00:00:00 2001 From: Roman Ettlinger Date: Mon, 19 Oct 2026 09:34:51 +0200 Subject: [PATCH] Server: index event duplicates only for queues that own their list Review feedback on the duplicate index: a derived queue that replaces an entry of m_events in place keeps its count, so the count check could not notice the change; SetQueueSize invalidated the index without clearing it, so a resized queue drained without another duplicate check kept up to 1000 published event instances alive; and the mirroring snapshot copied the queued range twice. The index is now used only by EventMonitoredItemQueue itself and the mirroring redundancy queue, which change the list only through the base members; any other derived queue scans the newest entries as before. SetQueueSize rebuilds the index right away, and the snapshot copies the range once. Co-Authored-By: Claude Opus 5.5 --- .../MirroringEventMonitoredItemQueue.cs | 4 ++ .../Queue/EventMonitoredItemQueue.cs | 59 +++++++++++-------- .../MonitoredItemQueueHandlerTests.cs | 46 +++++++++++++++ 3 files changed, 85 insertions(+), 24 deletions(-) diff --git a/src/Opc.Ua.Redundancy.Server/Subscriptions/MirroringEventMonitoredItemQueue.cs b/src/Opc.Ua.Redundancy.Server/Subscriptions/MirroringEventMonitoredItemQueue.cs index fdde5a6d25..4211966306 100644 --- a/src/Opc.Ua.Redundancy.Server/Subscriptions/MirroringEventMonitoredItemQueue.cs +++ b/src/Opc.Ua.Redundancy.Server/Subscriptions/MirroringEventMonitoredItemQueue.cs @@ -59,6 +59,10 @@ namespace Opc.Ua.Redundancy.Server m_sink = sink; } + /// + /// The queue changes its events only through the base members. + internal override bool IndexesRecentEvents => true; + /// public override void Enqueue(EventFieldList value) { diff --git a/src/Opc.Ua.Server/Subscription/MonitoredItem/Queue/EventMonitoredItemQueue.cs b/src/Opc.Ua.Server/Subscription/MonitoredItem/Queue/EventMonitoredItemQueue.cs index 068611137b..4853b206ed 100644 --- a/src/Opc.Ua.Server/Subscription/MonitoredItem/Queue/EventMonitoredItemQueue.cs +++ b/src/Opc.Ua.Server/Subscription/MonitoredItem/Queue/EventMonitoredItemQueue.cs @@ -86,8 +86,7 @@ namespace Opc.Ua.Server // advance the head instead of shifting the list: removing the first entry // moved every queued event for each dequeued one. value = m_events[m_head]; - bool indexed = m_indexedCount == m_events.Count; - if (indexed && ItemsInQueue <= kMaxNoOfEntriesCheckedForDuplicateEvents) + if (IndexesRecentEvents && ItemsInQueue <= kMaxNoOfEntriesCheckedForDuplicateEvents) { RemoveRecentHandle(value); } @@ -102,7 +101,6 @@ namespace Opc.Ua.Server // moves fewer events than were dequeued since the last compaction. CompactEvents(); } - m_indexedCount = indexed ? m_events.Count : -1; return true; } return false; @@ -138,11 +136,9 @@ namespace Opc.Ua.Server Dequeue(out EventFieldList _); } - bool indexed = m_indexedCount == m_events.Count; m_events.Add(value); - m_indexedCount = indexed ? m_events.Count : -1; - if (indexed) + if (IndexesRecentEvents) { AddRecentHandle(value); if (ItemsInQueue > kMaxNoOfEntriesCheckedForDuplicateEvents) @@ -161,13 +157,34 @@ namespace Opc.Ua.Server // path, so it sits among the most recently queued events. The handles of the // newest entries are counted in an index, so the check does not visit every // queued event for each event that is not a duplicate. - if (m_indexedCount != m_events.Count) + if (IndexesRecentEvents) { - RebuildRecentHandles(); + return instance != null && m_recentHandles != null && m_recentHandles.ContainsKey(instance); } - return instance != null && m_recentHandles != null && m_recentHandles.ContainsKey(instance); + + // a derived queue may change m_events directly, which the index cannot follow. + int lowestIndex = + ItemsInQueue > kMaxNoOfEntriesCheckedForDuplicateEvents + ? m_events.Count - (int)kMaxNoOfEntriesCheckedForDuplicateEvents + : m_head; + + for (int i = m_events.Count - 1; i >= lowestIndex; i--) + { + if (IsEventRaisedFor(m_events[i], instance)) + { + return true; + } + } + return false; } + /// + /// True when the duplicate check uses the index of the newest queued events. Only + /// queues that change solely through the members of this class + /// may use it; any other derived queue scans the newest entries instead. + /// + internal virtual bool IndexesRecentEvents => GetType() == typeof(EventMonitoredItemQueue); + /// /// Returns true when the queued event was raised for , for /// the duplicate check of @@ -206,8 +223,10 @@ namespace Opc.Ua.Server } } - // the next duplicate check counts the remaining events again. - m_indexedCount = -1; + if (IndexesRecentEvents) + { + RebuildRecentHandles(); + } } /// @@ -215,7 +234,9 @@ namespace Opc.Ua.Server /// internal EventFieldList[] CopyQueuedEvents() { - return m_events.GetRange(m_head, ItemsInQueue).ToArray(); + var events = new EventFieldList[ItemsInQueue]; + m_events.CopyTo(m_head, events, 0, events.Length); + return events; } /// @@ -225,16 +246,14 @@ namespace Opc.Ua.Server { if (m_head > 0) { - bool indexed = m_indexedCount == m_events.Count; m_events.RemoveRange(0, m_head); m_head = 0; - m_indexedCount = indexed ? m_events.Count : -1; } } /// - /// Counts the handles of the newest queued events again, after the queue was - /// truncated or was changed outside of this class. + /// Counts the handles of the newest queued events again after the queue was truncated, + /// and drops the handles of the events that left the queue. /// private void RebuildRecentHandles() { @@ -245,7 +264,6 @@ namespace Opc.Ua.Server { AddRecentHandle(m_events[i]); } - m_indexedCount = count; } /// @@ -326,13 +344,6 @@ namespace Opc.Ua.Server /// private Dictionary? m_recentHandles; - /// - /// The of that - /// matches; any other count makes the duplicate check - /// rebuild the index. - /// - private int m_indexedCount; - /// /// The logger to use here and in derived classes /// diff --git a/tests/Opc.Ua.Server.Tests/MonitoredItemQueueHandlerTests.cs b/tests/Opc.Ua.Server.Tests/MonitoredItemQueueHandlerTests.cs index 5a15953062..f6bb7c8949 100644 --- a/tests/Opc.Ua.Server.Tests/MonitoredItemQueueHandlerTests.cs +++ b/tests/Opc.Ua.Server.Tests/MonitoredItemQueueHandlerTests.cs @@ -27,8 +27,10 @@ * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ +using System; using System.Collections.Generic; using System.Linq; +using System.Runtime.CompilerServices; using Moq; using NUnit.Framework; using Opc.Ua.Tests; @@ -150,6 +152,45 @@ namespace Opc.Ua.Server.Tests Assert.That(queue.IsEventContainedInQueue(queued), Is.True); Assert.That(queue.Dequeue(out _), Is.True); Assert.That(queue.IsEventContainedInQueue(added), Is.False); + + // an entry replaced in place leaves the count unchanged. + var replacement = new AuditSessionEventState(null); + queue.ReplaceNewest(new EventFieldList { EventFields = [new Variant(5)], Handle = replacement }); + Assert.That(queue.IsEventContainedInQueue(replacement), Is.True); + Assert.That(queue.IsEventContainedInQueue(queued), Is.False); + } + + [Test] + public void EventQueueReleasesEventsAfterResizeAndDrain() + { + ITelemetryContext telemetry = NUnitTelemetryContext.Create(); + using var queue = new EventMonitoredItemQueue(false, 1, telemetry); + queue.SetQueueSize(10, true); + WeakReference released = EnqueueUnreferencedEvent(queue); + + queue.SetQueueSize(10, true); + DequeueAndDrop(queue); + Assert.That(queue.ItemsInQueue, Is.Zero); + + GC.Collect(); + GC.WaitForPendingFinalizers(); + GC.Collect(); + Assert.That(released.IsAlive, Is.False); + } + + [MethodImpl(MethodImplOptions.NoInlining)] + private static WeakReference EnqueueUnreferencedEvent(EventMonitoredItemQueue queue) + { + var handle = new AuditSessionEventState(null); + queue.Enqueue(new EventFieldList { EventFields = [new Variant(1)], Handle = handle }); + Assert.That(queue.IsEventContainedInQueue(handle), Is.True); + return new WeakReference(handle); + } + + [MethodImpl(MethodImplOptions.NoInlining)] + private static void DequeueAndDrop(EventMonitoredItemQueue queue) + { + Assert.That(queue.Dequeue(out _), Is.True); } private sealed class DirectlyFilledEventQueue : EventMonitoredItemQueue @@ -163,6 +204,11 @@ namespace Opc.Ua.Server.Tests { m_events.Add(value); } + + public void ReplaceNewest(EventFieldList value) + { + m_events[m_events.Count - 1] = value; + } } [TestCase(true)]