Merge #870 ring-buffer event monitored-item queue
Claude-Session: https://claude.ai/code/session_01GT67AD99zQQqpLCVm3p7xk
This commit is contained in:
Joseph Doherty
@@ -217,7 +217,7 @@ Two things to know before relying on any of this:
|
||||
replay off `Lock` one event per hold of a per-item gate, and withholds older deferred events from THAT item only
|
||||
(re-sent condition, or Retain=true of one omitted because NOT retained — never one omitted by ACL). Item state is a
|
||||
`ConditionalWeakTable` keyed by the item; cuts clear when the deferral count hits 0. Tests:
|
||||
`ConditionRefreshDeferralWireTests`. ⚠️ **Since #855 folder items are delivered per item too:** at every notifier we own, `DeliverConditionEventAsync` serves the `MonitoredNode2`'s attached items itself (`DeliverToNotifierItems`: audit guard → `INodeManager3.ValidateEventRolePermissions` per event, UNCACHED → item gate/cut → `QueueEvent`), never through the SDK channel — so no notifier of ours defers, a flood under `Lock` defers nothing, and folder refreshes (`RefreshDirectItems`, flag on AND off) share the #854 cut. The deferral queue + cut remain as the safety net; an event whose walk suspends is marked deferred BEFORE suspending. Tests produce deferral with `NotifierStall` (the `NotifierDeliveryStall` probe) — holding a folder item's SDK lock now blocks the reporter. `FolderConditionRefreshWireTests`; `docs/plans/2026-11-11-folder-event-per-item-delivery.md`. Folder events follow Part 3's shape —
|
||||
`ConditionRefreshDeferralWireTests`. ⚠️ **Since #855 folder items are delivered per item too:** at every notifier we own, `DeliverConditionEventAsync` serves the `MonitoredNode2`'s attached items itself (`DeliverToNotifierItems`: audit guard → `INodeManager3.ValidateEventRolePermissions` per event, UNCACHED → item gate/cut → `QueueEvent`), never through the SDK channel — so no notifier of ours defers, a flood under `Lock` defers nothing, and folder refreshes (`RefreshDirectItems`, flag on AND off) share the #854 cut. The deferral queue + cut remain as the safety net; an event whose walk suspends is marked deferred BEFORE suspending. Tests produce deferral with `NotifierStall` (the `NotifierDeliveryStall` probe) — holding a folder item's SDK lock now blocks the reporter. `FolderConditionRefreshWireTests`; `docs/plans/2026-11-11-folder-event-per-item-delivery.md`. Because that queueing runs under `Lock`, event item queues are `RingEventMonitoredItemQueue` (#870, supplied by `OtOpcUaSdkServer.CreateMonitoredItemQueueFactory` via `RingEventQueueFactory`; data-change queues stay the SDK's) — the SDK queue's discard-oldest was a `List.RemoveAt(0)` per stalled item per report. It is storage only (overflow events + DiscardOldest live in the SDK's unchanged `EventQueueHandler`); parity pinned by `RingEventMonitoredItemQueueParityTests` — re-check them on an SDK bump. Folder events follow Part 3's shape —
|
||||
`ReceiveEvents` on EventType + SourceNode (SDK check, kept first) AND our decision; non-condition events on a
|
||||
folder item fail closed under enforcement. Tests: `FolderEventFloodWireTests`. Overflow WITH the ACL doing work (#838
|
||||
item 2) — denied condition interleaved, folder + `i=2253` items, non-publishing client, concurrent Read, grant revoked
|
||||
|
||||
@@ -314,6 +314,27 @@ public sealed class OtOpcUaSdkServer : StandardServer
|
||||
_otOpcUaNodeManager!.InstallApplicationAuditPermissions(auditType);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Test/measurement switch (Gitea #870): false makes servers started afterwards use the SDK's own event queues, so
|
||||
/// the perf harness can A/B the two in one process. Never cleared in production.
|
||||
/// </summary>
|
||||
internal static bool RingEventQueues { get; set; } = true;
|
||||
|
||||
/// <summary>
|
||||
/// The SDK's factory with event queues swapped for <see cref="RingEventMonitoredItemQueue"/> (Gitea #870: since #855
|
||||
/// folder event items are queued under <c>Lock</c>, and the SDK queue's discard-oldest is an O(QueueSize) shift).
|
||||
/// A factory set through <see cref="StandardServer.MonitoredItemQueueFactory"/> is used as-is.
|
||||
/// </summary>
|
||||
/// <inheritdoc />
|
||||
protected override IMonitoredItemQueueFactory? CreateMonitoredItemQueueFactory(
|
||||
IServerInternal server, ApplicationConfiguration configuration)
|
||||
{
|
||||
var factory = base.CreateMonitoredItemQueueFactory(server, configuration);
|
||||
if (factory is null || ReferenceEquals(factory, MonitoredItemQueueFactory) || !RingEventQueues) return factory;
|
||||
// Server-owned (OwnsMonitoredItemQueueFactory: not the property's instance), so disposed on stop with the SDK's.
|
||||
return new RingEventQueueFactory(factory);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
protected override MasterNodeManager CreateMasterNodeManager(
|
||||
IServerInternal server, ApplicationConfiguration configuration)
|
||||
|
||||
@@ -0,0 +1,182 @@
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Server;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer;
|
||||
|
||||
/// <summary>
|
||||
/// A ring-buffer replacement for the SDK's <see cref="EventMonitoredItemQueue"/> (Gitea #870): the same observable
|
||||
/// behaviour, but discarding the oldest event at a full queue is O(1) instead of a <c>List.RemoveAt(0)</c> shift of
|
||||
/// the whole queue.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <b>Why.</b> Since #855 condition events are queued to folder event items by the reporter, under the node
|
||||
/// manager's <c>Lock</c>. At a full queue with DiscardOldest the SDK's <see cref="EventQueueHandler"/> dequeues the
|
||||
/// head and enqueues the new event; the SDK queue's dequeue is <c>List.RemoveAt(0)</c>, ~1 µs per item at
|
||||
/// QueueSize 1000 and ~10 µs at 10 000 — paid per stalled item per report while <c>Lock</c> is held.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// <b>What is reproduced (SDK 2.0.1, decompiled).</b> The SDK queue is pure storage — DiscardOldest, the
|
||||
/// <c>EventQueueOverflowEventType</c> positions and the discarded/overflow counters all live in
|
||||
/// <see cref="EventQueueHandler"/>, which is unchanged and drives this queue exactly as it drives the SDK's.
|
||||
/// Reproduced here: <see cref="QueueSize"/> starts at 0; <see cref="Enqueue"/> at QueueSize 0 throws
|
||||
/// <see cref="ServiceResultException"/> <c>BadInternalError</c>, and at <c>Count == QueueSize</c> drops the oldest
|
||||
/// entry first; <see cref="Dequeue"/> is FIFO and returns false (value null) when empty;
|
||||
/// <see cref="SetQueueSize"/> keeps the newest <c>queueSize</c> entries when <c>discardOldest</c>, else the
|
||||
/// oldest; <see cref="IsEventContainedInQueue"/> checks the newest ≤ 1000 entries with the SDK's own
|
||||
/// <see cref="EventMonitoredItemQueue.IsEventRaisedFor"/> (null entries skipped); <see cref="IsDurable"/> is
|
||||
/// false; <see cref="Dispose()"/> does nothing observable (the SDK's does nothing at all). A durable queue is
|
||||
/// never created here — <see cref="RingEventQueueFactory"/> hands that request to the SDK factory, which refuses it
|
||||
/// exactly as before.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// Not thread-safe, like the SDK's: <see cref="EventQueueHandler"/> is only used under its monitored item's lock.
|
||||
/// Parity is pinned by <c>RingEventMonitoredItemQueueParityTests</c>.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class RingEventMonitoredItemQueue : IEventMonitoredItemQueue
|
||||
{
|
||||
/// <summary>The SDK's <c>kMaxNoOfEntriesCheckedForDuplicateEvents</c>.</summary>
|
||||
internal const int MaxEntriesCheckedForDuplicates = 1000;
|
||||
|
||||
private const int MinimumCapacity = 4;
|
||||
|
||||
private EventFieldList?[] _buffer = [];
|
||||
private int _head;
|
||||
private int _count;
|
||||
|
||||
/// <summary>Creates an empty queue (QueueSize 0, as the SDK's).</summary>
|
||||
/// <param name="monitoredItemId">The monitored item the queue belongs to.</param>
|
||||
public RingEventMonitoredItemQueue(uint monitoredItemId) => MonitoredItemId = monitoredItemId;
|
||||
|
||||
/// <inheritdoc />
|
||||
public uint MonitoredItemId { get; }
|
||||
|
||||
/// <inheritdoc />
|
||||
public bool IsDurable => false;
|
||||
|
||||
/// <inheritdoc />
|
||||
public uint QueueSize { get; private set; }
|
||||
|
||||
/// <inheritdoc />
|
||||
public int ItemsInQueue => _count;
|
||||
|
||||
/// <summary>Allocated slots (test observation: the buffer grows with use and shrinks with the queue size).</summary>
|
||||
internal int Capacity => _buffer.Length;
|
||||
|
||||
/// <inheritdoc />
|
||||
public void SetQueueSize(uint queueSize, bool discardOldest)
|
||||
{
|
||||
QueueSize = queueSize;
|
||||
if (_count > queueSize)
|
||||
{
|
||||
var excess = _count - (int)queueSize;
|
||||
if (discardOldest)
|
||||
{
|
||||
for (var i = 0; i < excess; i++) DropHead();
|
||||
}
|
||||
else
|
||||
{
|
||||
for (var i = 0; i < excess; i++)
|
||||
{
|
||||
_buffer[Slot(_count - 1)] = null;
|
||||
_count--;
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
// Not observable: release a buffer much larger than the new size (a List would keep it).
|
||||
if (_buffer.Length > MinimumCapacity && _buffer.Length > 2 * Math.Max(queueSize, (uint)MinimumCapacity))
|
||||
Resize(Math.Max(_count, MinimumCapacity));
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public bool IsEventContainedInQueue(IFilterTarget instance)
|
||||
{
|
||||
var oldest = _count > MaxEntriesCheckedForDuplicates ? _count - MaxEntriesCheckedForDuplicates : 0;
|
||||
for (var i = _count - 1; i >= oldest; i--)
|
||||
{
|
||||
var queued = _buffer[Slot(i)];
|
||||
if (queued is not null && EventMonitoredItemQueue.IsEventRaisedFor(queued, instance)) return true;
|
||||
}
|
||||
|
||||
return false;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Enqueue(EventFieldList value)
|
||||
{
|
||||
if (QueueSize == 0)
|
||||
throw new ServiceResultException(StatusCodes.BadInternalError, "Error queueing Event. Queue size is set to 0");
|
||||
|
||||
if (_count == QueueSize) DropHead();
|
||||
|
||||
if (_count == _buffer.Length)
|
||||
Resize((int)Math.Min(QueueSize, (ulong)Math.Max(MinimumCapacity, _buffer.Length * 2L)));
|
||||
|
||||
_buffer[Slot(_count)] = value;
|
||||
_count++;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public bool Dequeue(out EventFieldList value)
|
||||
{
|
||||
value = null!;
|
||||
if (_count == 0) return false;
|
||||
value = _buffer[_head]!;
|
||||
DropHead();
|
||||
return true;
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Dispose()
|
||||
{
|
||||
// The SDK's queue disposes nothing; the queue stays usable, as the SDK's does.
|
||||
}
|
||||
|
||||
private int Slot(int index)
|
||||
{
|
||||
var slot = _head + index;
|
||||
return slot >= _buffer.Length ? slot - _buffer.Length : slot;
|
||||
}
|
||||
|
||||
private void DropHead()
|
||||
{
|
||||
_buffer[_head] = null;
|
||||
_head = _head + 1 == _buffer.Length ? 0 : _head + 1;
|
||||
_count--;
|
||||
if (_count == 0) _head = 0;
|
||||
}
|
||||
|
||||
private void Resize(int capacity)
|
||||
{
|
||||
var next = new EventFieldList?[Math.Max(capacity, _count)];
|
||||
for (var i = 0; i < _count; i++) next[i] = _buffer[Slot(i)];
|
||||
_buffer = next;
|
||||
_head = 0;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The server's <see cref="IMonitoredItemQueueFactory"/> (Gitea #870): event queues are
|
||||
/// <see cref="RingEventMonitoredItemQueue"/>; data-change queues, durable requests and disposal go to the SDK's own
|
||||
/// factory unchanged.
|
||||
/// </summary>
|
||||
public sealed class RingEventQueueFactory(IMonitoredItemQueueFactory inner) : IMonitoredItemQueueFactory
|
||||
{
|
||||
private readonly IMonitoredItemQueueFactory _inner = inner ?? throw new ArgumentNullException(nameof(inner));
|
||||
|
||||
/// <inheritdoc />
|
||||
public bool SupportsDurableQueues => _inner.SupportsDurableQueues;
|
||||
|
||||
/// <inheritdoc />
|
||||
public IDataChangeMonitoredItemQueue CreateDataChangeQueue(bool isDurable, uint monitoredItemId) =>
|
||||
_inner.CreateDataChangeQueue(isDurable, monitoredItemId);
|
||||
|
||||
/// <inheritdoc />
|
||||
public IEventMonitoredItemQueue CreateEventQueue(bool isDurable, uint monitoredItemId) =>
|
||||
isDurable ? _inner.CreateEventQueue(isDurable, monitoredItemId) : new RingEventMonitoredItemQueue(monitoredItemId);
|
||||
|
||||
/// <inheritdoc />
|
||||
public void Dispose() => _inner.Dispose();
|
||||
}
|
||||
+37
-2
@@ -35,6 +35,9 @@ namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
/// compiles against the pre-#855 tree for a before/after comparison.</para>
|
||||
/// </remarks>
|
||||
[Trait("Category", "Perf")]
|
||||
// Shares a collection with the other user of OtOpcUaSdkServer.RingEventQueues (a process-wide switch the perf
|
||||
// harness flips), so the wiring assertion never observes it switched off (#870 review).
|
||||
[Collection(RingEventQueueSwitchCollection.Name)]
|
||||
public sealed class FolderEventDeliveryPerfTests
|
||||
{
|
||||
private const string OptInVariable = "OTOPCUA_EVENT_PERF";
|
||||
@@ -100,8 +103,40 @@ public sealed class FolderEventDeliveryPerfTests
|
||||
public Task Stalled_items_queue_size_sensitivity(uint queueSize) =>
|
||||
RunAsync(50, acl: false, stalled: true, serverObjectProbe: false, queueSize);
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #870 A/B in one process (same machine load): N stalled folder items at a full queue, with the server's
|
||||
/// ring-buffer event queues vs the SDK's own <c>EventMonitoredItemQueue</c> (<c>List.RemoveAt(0)</c> per discard).
|
||||
/// Each round runs both arms, alternating which goes first, so a load spike lands on both.
|
||||
/// <c>OTOPCUA_EVENT_PERF_ROUNDS</c> sets the rounds (default 2).
|
||||
/// </summary>
|
||||
[Theory]
|
||||
[InlineData(50, 1000u, false)]
|
||||
[InlineData(200, 1000u, false)]
|
||||
[InlineData(200, 1000u, true)]
|
||||
[InlineData(50, 10000u, false)]
|
||||
public async Task Ring_vs_sdk_event_queue(int n, uint queueSize, bool acl)
|
||||
{
|
||||
RequireOptIn();
|
||||
var rounds = int.TryParse(Environment.GetEnvironmentVariable("OTOPCUA_EVENT_PERF_ROUNDS"), out var r) && r > 0 ? r : 2;
|
||||
try
|
||||
{
|
||||
for (var round = 0; round < rounds; round++)
|
||||
{
|
||||
foreach (var ring in round % 2 == 0 ? new[] { false, true } : [true, false])
|
||||
{
|
||||
OtOpcUaSdkServer.RingEventQueues = ring;
|
||||
await RunAsync(n, acl, stalled: true, serverObjectProbe: false, queueSize, ring ? "ring" : "sdk ");
|
||||
}
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
OtOpcUaSdkServer.RingEventQueues = true;
|
||||
}
|
||||
}
|
||||
|
||||
private static async Task RunAsync(
|
||||
int n, bool acl, bool stalled, bool serverObjectProbe, uint queueSize = DefaultQueueSize)
|
||||
int n, bool acl, bool stalled, bool serverObjectProbe, uint queueSize = DefaultQueueSize, string? queueKind = null)
|
||||
{
|
||||
RequireOptIn();
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
@@ -195,7 +230,7 @@ public sealed class FolderEventDeliveryPerfTests
|
||||
var deferredPeak = rig.Nm.DeferredEventReports;
|
||||
var row = new StringBuilder();
|
||||
row.Append(CultureInfo.InvariantCulture,
|
||||
$"{(serverObjectProbe ? "i=2253" : "folder"),-6} N={n,-3} acl={(acl ? "on " : "off")} {(stalled ? "stalled " : "publishing")} q={queueSize,-5} | ");
|
||||
$"{(serverObjectProbe ? "i=2253" : "folder"),-6} N={n,-3} acl={(acl ? "on " : "off")} {(stalled ? "stalled " : "publishing")} q={queueSize,-5} {(queueKind is null ? "" : queueKind + " ")}| ");
|
||||
row.Append(CultureInfo.InvariantCulture, $"lock hold us p50 {Pct(holds, 50),8:F1} p90 {Pct(holds, 90),8:F1} p99 {Pct(holds, 99),8:F1} max {holds.Max(),9:F1} mean {holds.Average(),8:F1}");
|
||||
row.Append(CultureInfo.InvariantCulture, $" | per-item mean {(n > 0 ? holds.Average() / n : 0),6:F2} us");
|
||||
if (expectFolder)
|
||||
|
||||
+36
@@ -0,0 +1,36 @@
|
||||
using Opc.Ua.Server;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.IntegrationTests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #870: the running server's monitored-item queue factory is <see cref="RingEventQueueFactory"/> (every event
|
||||
/// item's queue is a <see cref="RingEventMonitoredItemQueue"/>; the SDK's <see cref="EventQueueHandler"/> always asks
|
||||
/// the server's factory), and data-change queues are still the SDK's own.
|
||||
/// </summary>
|
||||
// Shares a collection with the other user of OtOpcUaSdkServer.RingEventQueues (a process-wide switch the perf
|
||||
// harness flips), so the wiring assertion never observes it switched off (#870 review).
|
||||
[Collection(RingEventQueueSwitchCollection.Name)]
|
||||
public sealed class RingEventQueueWiringTests
|
||||
{
|
||||
[Fact]
|
||||
public async Task The_server_creates_ring_event_queues_and_sdk_data_change_queues()
|
||||
{
|
||||
var ct = TestContext.Current.CancellationToken;
|
||||
await using var rig = await ServerEventRig.StartAsync("RingQueue870", encrypted: false, ct);
|
||||
|
||||
var factory = rig.Server.CurrentInstance.MonitoredItemQueueFactory.ShouldBeOfType<RingEventQueueFactory>();
|
||||
factory.SupportsDurableQueues.ShouldBeFalse();
|
||||
factory.CreateEventQueue(false, 1).ShouldBeOfType<RingEventMonitoredItemQueue>();
|
||||
factory.CreateDataChangeQueue(false, 2).ShouldBeOfType<DataChangeMonitoredItemQueue>();
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Serialises the tests that read or flip <c>OtOpcUaSdkServer.RingEventQueues</c>.</summary>
|
||||
[CollectionDefinition(Name)]
|
||||
public sealed class RingEventQueueSwitchCollection
|
||||
{
|
||||
/// <summary>The collection name.</summary>
|
||||
public const string Name = "RingEventQueueSwitch";
|
||||
}
|
||||
+308
@@ -0,0 +1,308 @@
|
||||
using Opc.Ua;
|
||||
using Opc.Ua.Server;
|
||||
using Shouldly;
|
||||
using Xunit;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.OpcUaServer.Tests;
|
||||
|
||||
/// <summary>
|
||||
/// Gitea #870: <see cref="RingEventMonitoredItemQueue"/> must be observably identical to the SDK's
|
||||
/// <see cref="EventMonitoredItemQueue"/>. Every test drives the SDK queue and the ring with the same operations and
|
||||
/// compares every result — directly at the queue, and through the SDK's own <see cref="EventQueueHandler"/> (which owns
|
||||
/// DiscardOldest, the overflow-event positions and the discarded/overflow counters) built over each factory.
|
||||
/// </summary>
|
||||
public sealed class RingEventMonitoredItemQueueParityTests
|
||||
{
|
||||
private static readonly ITelemetryContext Telemetry = DefaultTelemetry.Create(_ => { });
|
||||
|
||||
private sealed class Target : IFilterTarget
|
||||
{
|
||||
public bool IsTypeOf(IFilterContext context, NodeId typeDefinitionId) => false;
|
||||
|
||||
public Variant GetAttributeValue(IFilterContext context, NodeId typeDefinitionId,
|
||||
ArrayOf<QualifiedName> relativePath, uint attributeId, NumericRange indexRange) => Variant.Null;
|
||||
}
|
||||
|
||||
private static readonly Target[] Targets = Enumerable.Range(0, 6).Select(_ => new Target()).ToArray();
|
||||
|
||||
private static EventFieldList NewEvent(Random random) => new() { Handle = Targets[random.Next(Targets.Length)] };
|
||||
|
||||
/// <summary>Runs <paramref name="action"/> on both, asserting the same outcome (value or exception type + status).</summary>
|
||||
private static T Both<T>(Func<IEventMonitoredItemQueue, T> action, IEventMonitoredItemQueue sdk,
|
||||
IEventMonitoredItemQueue ring, string step)
|
||||
{
|
||||
Exception? sdkError = null, ringError = null;
|
||||
T sdkResult = default!, ringResult = default!;
|
||||
try { sdkResult = action(sdk); } catch (Exception e) { sdkError = e; }
|
||||
try { ringResult = action(ring); } catch (Exception e) { ringError = e; }
|
||||
|
||||
(ringError?.GetType()).ShouldBe(sdkError?.GetType(), step);
|
||||
if (sdkError is ServiceResultException s)
|
||||
((ServiceResultException)ringError!).StatusCode.ShouldBe(s.StatusCode, step);
|
||||
if (sdkError is null) ringResult.ShouldBe(sdkResult, step);
|
||||
ring.ItemsInQueue.ShouldBe(sdk.ItemsInQueue, step);
|
||||
ring.QueueSize.ShouldBe(sdk.QueueSize, step);
|
||||
return sdkResult;
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Initial_state_matches()
|
||||
{
|
||||
using var sdk = new EventMonitoredItemQueue(false, 7, Telemetry);
|
||||
using var ring = new RingEventMonitoredItemQueue(7);
|
||||
ring.MonitoredItemId.ShouldBe(sdk.MonitoredItemId);
|
||||
ring.IsDurable.ShouldBe(sdk.IsDurable);
|
||||
ring.QueueSize.ShouldBe(sdk.QueueSize);
|
||||
ring.ItemsInQueue.ShouldBe(sdk.ItemsInQueue);
|
||||
ring.Dequeue(out var value).ShouldBeFalse();
|
||||
value.ShouldBeNull();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Enqueue_at_queue_size_zero_throws_the_same_BadInternalError()
|
||||
{
|
||||
var sdk = new EventMonitoredItemQueue(false, 1, Telemetry);
|
||||
var ring = new RingEventMonitoredItemQueue(1);
|
||||
Both(q => { q.Enqueue(new EventFieldList()); return 0; }, sdk, ring, "enqueue at 0");
|
||||
Should.Throw<ServiceResultException>(() => ring.Enqueue(new EventFieldList())).StatusCode
|
||||
.ShouldBe(StatusCodes.BadInternalError);
|
||||
}
|
||||
|
||||
[Theory]
|
||||
[InlineData(true)]
|
||||
[InlineData(false)]
|
||||
public void Shrink_keeps_newest_or_oldest_and_dequeue_order_is_fifo(bool discardOldest)
|
||||
{
|
||||
var sdk = new EventMonitoredItemQueue(false, 1, Telemetry);
|
||||
var ring = new RingEventMonitoredItemQueue(1);
|
||||
var events = Enumerable.Range(0, 10).Select(_ => new EventFieldList()).ToArray();
|
||||
sdk.SetQueueSize(10, true);
|
||||
ring.SetQueueSize(10, true);
|
||||
// Wrap the ring first so the shrink crosses the buffer end.
|
||||
for (var i = 0; i < 6; i++) { sdk.Enqueue(new EventFieldList()); ring.Enqueue(new EventFieldList()); }
|
||||
for (var i = 0; i < 6; i++) { sdk.Dequeue(out _); ring.Dequeue(out _); }
|
||||
foreach (var e in events) { sdk.Enqueue(e); ring.Enqueue(e); }
|
||||
|
||||
sdk.SetQueueSize(4, discardOldest);
|
||||
ring.SetQueueSize(4, discardOldest);
|
||||
|
||||
var expected = discardOldest ? events[6..] : events[..4];
|
||||
foreach (var e in expected)
|
||||
{
|
||||
sdk.Dequeue(out var a).ShouldBeTrue();
|
||||
ring.Dequeue(out var b).ShouldBeTrue();
|
||||
a.ShouldBeSameAs(e);
|
||||
b.ShouldBeSameAs(e);
|
||||
}
|
||||
|
||||
ring.Dequeue(out _).ShouldBeFalse();
|
||||
sdk.Dequeue(out _).ShouldBeFalse();
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Enqueue_at_a_full_queue_drops_the_oldest()
|
||||
{
|
||||
var ring = new RingEventMonitoredItemQueue(1);
|
||||
var sdk = new EventMonitoredItemQueue(false, 1, Telemetry);
|
||||
ring.SetQueueSize(3, false);
|
||||
sdk.SetQueueSize(3, false);
|
||||
var events = Enumerable.Range(0, 5).Select(_ => new EventFieldList()).ToArray();
|
||||
foreach (var e in events) { ring.Enqueue(e); sdk.Enqueue(e); }
|
||||
foreach (var e in events[2..])
|
||||
{
|
||||
ring.Dequeue(out var b).ShouldBeTrue();
|
||||
sdk.Dequeue(out var a).ShouldBeTrue();
|
||||
b.ShouldBeSameAs(e);
|
||||
a.ShouldBeSameAs(e);
|
||||
}
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Duplicate_check_covers_exactly_the_newest_1000_entries()
|
||||
{
|
||||
var sdk = new EventMonitoredItemQueue(false, 1, Telemetry);
|
||||
var ring = new RingEventMonitoredItemQueue(1);
|
||||
sdk.SetQueueSize(3000, true);
|
||||
ring.SetQueueSize(3000, true);
|
||||
var old = new Target();
|
||||
var edge = new Target();
|
||||
void Add(EventFieldList e) { sdk.Enqueue(e); ring.Enqueue(e); }
|
||||
Add(new EventFieldList { Handle = old });
|
||||
Add(new EventFieldList { Handle = edge });
|
||||
for (var i = 0; i < 999; i++) Add(new EventFieldList { Handle = Targets[0] });
|
||||
Add(null!); // a null entry is skipped, as the SDK skips it
|
||||
|
||||
// 1002 entries: `old` is the 1002nd newest, `edge` the 1001st — both outside the 1000-entry window.
|
||||
Both(q => q.IsEventContainedInQueue(old), sdk, ring, "old").ShouldBeFalse();
|
||||
Both(q => q.IsEventContainedInQueue(edge), sdk, ring, "edge").ShouldBeFalse();
|
||||
sdk.Dequeue(out _); ring.Dequeue(out _);
|
||||
sdk.Dequeue(out _); ring.Dequeue(out _);
|
||||
Add(new EventFieldList { Handle = edge });
|
||||
Both(q => q.IsEventContainedInQueue(edge), sdk, ring, "newest").ShouldBeTrue();
|
||||
Both(q => q.IsEventContainedInQueue(Targets[0]), sdk, ring, "middle").ShouldBeTrue();
|
||||
Both(q => q.IsEventContainedInQueue(Targets[1]), sdk, ring, "absent").ShouldBeFalse();
|
||||
}
|
||||
|
||||
/// <summary>Upper boundary of the window: the entry exactly 1000th from newest is still found; 1001st is not.</summary>
|
||||
[Fact]
|
||||
public void Duplicate_check_includes_the_1000th_newest_entry()
|
||||
{
|
||||
var sdk = new EventMonitoredItemQueue(false, 1, Telemetry);
|
||||
var ring = new RingEventMonitoredItemQueue(1);
|
||||
sdk.SetQueueSize(3000, true);
|
||||
ring.SetQueueSize(3000, true);
|
||||
var outside = new Target();
|
||||
var boundary = new Target();
|
||||
void Add(EventFieldList e) { sdk.Enqueue(e); ring.Enqueue(e); }
|
||||
Add(new EventFieldList { Handle = outside });
|
||||
Add(new EventFieldList { Handle = boundary });
|
||||
for (var i = 0; i < 999; i++) Add(new EventFieldList { Handle = Targets[0] });
|
||||
|
||||
// 1001 entries: `boundary` is the 1000th newest (inside), `outside` the 1001st.
|
||||
Both(q => q.IsEventContainedInQueue(boundary), sdk, ring, "boundary").ShouldBeTrue();
|
||||
Both(q => q.IsEventContainedInQueue(outside), sdk, ring, "outside").ShouldBeFalse();
|
||||
}
|
||||
|
||||
/// <summary>Random interleavings of every queue operation, at small sizes (wrap-around, grow, shrink, size 0).</summary>
|
||||
[Theory]
|
||||
[MemberData(nameof(Seeds))]
|
||||
public void Random_operation_sequences_match_the_sdk_queue(int seed)
|
||||
{
|
||||
var random = new Random(seed);
|
||||
var sdk = new EventMonitoredItemQueue(false, 3, Telemetry);
|
||||
var ring = new RingEventMonitoredItemQueue(3);
|
||||
for (var step = 0; step < 2000; step++)
|
||||
{
|
||||
var label = $"seed {seed} step {step}";
|
||||
switch (random.Next(10))
|
||||
{
|
||||
case < 5:
|
||||
var e = random.Next(50) == 0 ? null! : NewEvent(random);
|
||||
Both(q => { q.Enqueue(e); return 0; }, sdk, ring, label);
|
||||
break;
|
||||
case < 8:
|
||||
var (sdkHad, sdkValue) = (sdk.Dequeue(out var a), a);
|
||||
var (ringHad, ringValue) = (ring.Dequeue(out var b), b);
|
||||
ringHad.ShouldBe(sdkHad, label);
|
||||
ringValue.ShouldBeSameAs(sdkValue, label);
|
||||
break;
|
||||
case 8:
|
||||
var size = (uint)(random.Next(8) == 0 ? 0 : random.Next(1, 40));
|
||||
var discardOldest = random.Next(2) == 0;
|
||||
Both(q => { q.SetQueueSize(size, discardOldest); return 0; }, sdk, ring, label);
|
||||
break;
|
||||
default:
|
||||
var t = Targets[random.Next(Targets.Length)];
|
||||
Both(q => q.IsEventContainedInQueue(t), sdk, ring, label);
|
||||
break;
|
||||
}
|
||||
}
|
||||
|
||||
// Drain: the same remaining contents in the same order.
|
||||
while (sdk.Dequeue(out var a))
|
||||
{
|
||||
ring.Dequeue(out var b).ShouldBeTrue();
|
||||
b.ShouldBeSameAs(a);
|
||||
}
|
||||
|
||||
ring.Dequeue(out _).ShouldBeFalse();
|
||||
}
|
||||
|
||||
public static TheoryData<int> Seeds() => new(Enumerable.Range(1, 40));
|
||||
|
||||
/// <summary>
|
||||
/// The SDK's <see cref="EventQueueHandler"/> over the SDK factory vs over <see cref="RingEventQueueFactory"/>:
|
||||
/// QueueEvent (incl. the DiscardOldest=false refusal), SetQueueOverflowIfFull, SetQueueSize with DiscardOldest
|
||||
/// flips, Publish with small batches (dequeue mid-stream) — the published events, the overflow-event-due flags,
|
||||
/// the Overflow property and the discarded/overflow callbacks must all agree.
|
||||
/// </summary>
|
||||
[Theory]
|
||||
[MemberData(nameof(Seeds))]
|
||||
public void The_sdk_event_queue_handler_behaves_identically_over_either_queue(int seed)
|
||||
{
|
||||
var random = new Random(seed);
|
||||
var counts = new int[4]; // sdk discarded, sdk overflow, ring discarded, ring overflow
|
||||
using var sdkFactory = new MonitoredItemQueueFactory(Telemetry);
|
||||
using var ringFactory = new RingEventQueueFactory(new MonitoredItemQueueFactory(Telemetry));
|
||||
using var sdk = new EventQueueHandler(false, sdkFactory, 9, Telemetry, () => counts[0]++, () => counts[1]++);
|
||||
using var ring = new EventQueueHandler(false, ringFactory, 9, Telemetry, () => counts[2]++, () => counts[3]++);
|
||||
var context = new OperationContext(new RequestHeader(), null, RequestType.Publish, RequestLifetime.None);
|
||||
|
||||
var initialOldest = random.Next(2) == 0;
|
||||
sdk.SetQueueSize(10, initialOldest);
|
||||
ring.SetQueueSize(10, initialOldest);
|
||||
|
||||
for (var step = 0; step < 1500; step++)
|
||||
{
|
||||
var label = $"seed {seed} step {step}";
|
||||
switch (random.Next(12))
|
||||
{
|
||||
case < 6:
|
||||
var e = NewEvent(random);
|
||||
Exception? a = null, b = null;
|
||||
try { sdk.QueueEvent(e); } catch (Exception x) { a = x; }
|
||||
try { ring.QueueEvent(e); } catch (Exception x) { b = x; }
|
||||
(b?.GetType()).ShouldBe(a?.GetType(), label);
|
||||
break;
|
||||
case 6:
|
||||
ring.SetQueueOverflowIfFull().ShouldBe(sdk.SetQueueOverflowIfFull(), label);
|
||||
break;
|
||||
case 7:
|
||||
var size = (uint)random.Next(1, 25);
|
||||
var discardOldest = random.Next(2) == 0;
|
||||
sdk.SetQueueSize(size, discardOldest);
|
||||
ring.SetQueueSize(size, discardOldest);
|
||||
break;
|
||||
case < 10:
|
||||
var max = (uint)random.Next(1, 6);
|
||||
var sdkOut = new Queue<EventFieldList>();
|
||||
var ringOut = new Queue<EventFieldList>();
|
||||
var sdkN = sdk.Publish(context, sdkOut, max);
|
||||
var ringN = ring.Publish(context, ringOut, max);
|
||||
ringN.ShouldBe(sdkN, label);
|
||||
ringOut.Count.ShouldBe(sdkOut.Count, label);
|
||||
while (sdkOut.Count > 0) ringOut.Dequeue().ShouldBeSameAs(sdkOut.Dequeue(), label);
|
||||
break;
|
||||
default:
|
||||
var t = Targets[random.Next(Targets.Length)];
|
||||
ring.IsEventContainedInQueue(t).ShouldBe(sdk.IsEventContainedInQueue(t), label);
|
||||
break;
|
||||
}
|
||||
|
||||
ring.ItemsInQueue.ShouldBe(sdk.ItemsInQueue, label);
|
||||
ring.Overflow.ShouldBe(sdk.Overflow, label);
|
||||
counts[2].ShouldBe(counts[0], label);
|
||||
counts[3].ShouldBe(counts[1], label);
|
||||
}
|
||||
|
||||
// Positive control: the sequence really overflowed (discards and overflow events happened).
|
||||
counts[0].ShouldBeGreaterThan(0);
|
||||
counts[1].ShouldBeGreaterThan(0);
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Factory_supplies_ring_event_queues_and_delegates_everything_else()
|
||||
{
|
||||
using var factory = new RingEventQueueFactory(new MonitoredItemQueueFactory(Telemetry));
|
||||
factory.SupportsDurableQueues.ShouldBeFalse();
|
||||
factory.CreateEventQueue(false, 4).ShouldBeOfType<RingEventMonitoredItemQueue>().MonitoredItemId.ShouldBe(4u);
|
||||
factory.CreateDataChangeQueue(false, 5).ShouldBeOfType<DataChangeMonitoredItemQueue>();
|
||||
// A durable request reaches the SDK factory, which refuses it as before.
|
||||
Should.Throw<ArgumentException>(() => factory.CreateEventQueue(true, 6));
|
||||
Should.Throw<ArgumentException>(() => new MonitoredItemQueueFactory(Telemetry).CreateEventQueue(true, 6));
|
||||
}
|
||||
|
||||
[Fact]
|
||||
public void Buffer_grows_lazily_and_is_released_on_shrink()
|
||||
{
|
||||
var ring = new RingEventMonitoredItemQueue(1);
|
||||
ring.SetQueueSize(10000, true);
|
||||
ring.Capacity.ShouldBe(0);
|
||||
for (var i = 0; i < 20000; i++) ring.Enqueue(new EventFieldList());
|
||||
ring.Capacity.ShouldBe(10000);
|
||||
ring.ItemsInQueue.ShouldBe(10000);
|
||||
ring.SetQueueSize(10, true);
|
||||
ring.ItemsInQueue.ShouldBe(10);
|
||||
ring.Capacity.ShouldBeLessThanOrEqualTo(20);
|
||||
}
|
||||
}
|
||||
Reference in New Issue
Block a user