GithubStar/UA-.NETStandard
1
0
Fork 0
mirror of https://github.com/OPCFoundation/UA-.NETStandard.git synced 2026-10-27 21:55:54 +08:00
Code Issues Packages Projects Releases Wiki Activity

Server: index event duplicates only for queues that own their list

Browse Source 
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 <noreply@anthropic.com>
This commit is contained in:
Roman Ettlinger
2026-10-19 09:34:51 +02:00
co-authored by Claude Opus 5.5
parent 7351b3b088
commit 27b6ad0d0a
3 changed files with 85 additions and 24 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/Opc.Ua.Redundancy.Server/Subscriptions/MirroringEventMonitoredItemQueue.cs
+4
View File
@@ -59,6 +59,10 @@ namespace Opc.Ua.Redundancy.Server
m_sink = sink;
}
/// <inheritdoc/>
/// <remarks>The queue changes its events only through the base members.</remarks>
internal override bool IndexesRecentEvents => true;
/// <inheritdoc/>
public override void Enqueue(EventFieldList value)
{
src/Opc.Ua.Server/Subscription/MonitoredItem/Queue/EventMonitoredItemQueue.cs
+35 -24
View File
@@ -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;
}
/// <summary>
/// True when the duplicate check uses the index of the newest queued events. Only
/// queues that change <see cref="m_events"/> solely through the members of this class
/// may use it; any other derived queue scans the newest entries instead.
/// </summary>
internal virtual bool IndexesRecentEvents => GetType() == typeof(EventMonitoredItemQueue);
/// <summary>
/// Returns true when the queued event was raised for <paramref name="instance"/>, for
/// the duplicate check of <see cref="IEventMonitoredItemQueue.IsEventContainedInQueue"/>
@@ -206,8 +223,10 @@ namespace Opc.Ua.Server
}
}
// the next duplicate check counts the remaining events again.
m_indexedCount = -1;
if (IndexesRecentEvents)
{
RebuildRecentHandles();
}
}
/// <summary>
@@ -215,7 +234,9 @@ namespace Opc.Ua.Server
/// </summary>
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;
}
/// <summary>
@@ -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;
}
}
/// <summary>
/// Counts the handles of the newest queued events again, after the queue was
/// truncated or <see cref="m_events"/> 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.
/// </summary>
private void RebuildRecentHandles()
{
@@ -245,7 +264,6 @@ namespace Opc.Ua.Server
{
AddRecentHandle(m_events[i]);
}
m_indexedCount = count;
}
/// <summary>
@@ -326,13 +344,6 @@ namespace Opc.Ua.Server
/// </summary>
private Dictionary<object, int>? m_recentHandles;
/// <summary>
/// The <see cref="List{T}.Count"/> of <see cref="m_events"/> that
/// <see cref="m_recentHandles"/> matches; any other count makes the duplicate check
/// rebuild the index.
/// </summary>
private int m_indexedCount;
/// <summary>
/// The logger to use here and in derived classes
/// </summary>
tests/Opc.Ua.Server.Tests/MonitoredItemQueueHandlerTests.cs
+46
View File
@@ -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)]