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)]