Joseph Doherty
732 lines
31 KiB
C#
732 lines
31 KiB
C#
using System;
|
|
using System.Collections.Generic;
|
|
using System.Diagnostics;
|
|
using System.Threading;
|
|
using System.Threading.Tasks;
|
|
using Google.Protobuf.WellKnownTypes;
|
|
using ZB.MOM.WW.MxGateway.Contracts.Proto;
|
|
using ZB.MOM.WW.MxGateway.Worker.MxAccess;
|
|
|
|
namespace ZB.MOM.WW.MxGateway.Worker.Tests.MxAccess;
|
|
|
|
public sealed class MxAccessEventQueueTests
|
|
{
|
|
/// <summary>Verifies that Enqueue assigns monotonic worker sequences and preserves event order.</summary>
|
|
[Fact]
|
|
public void Enqueue_AssignsMonotonicWorkerSequencesAndPreservesOrder()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 4);
|
|
|
|
queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 10));
|
|
queue.Enqueue(CreateEvent(MxEventFamily.OnWriteComplete, itemHandle: 11));
|
|
|
|
Assert.Equal(2, queue.Count);
|
|
Assert.Equal(2UL, queue.LastEventSequence);
|
|
|
|
Assert.True(queue.TryDequeue(out WorkerEvent? dequeuedFirst));
|
|
Assert.True(queue.TryDequeue(out WorkerEvent? dequeuedSecond));
|
|
Assert.Equal(1UL, dequeuedFirst?.Event.WorkerSequence);
|
|
Assert.Equal(2UL, dequeuedSecond?.Event.WorkerSequence);
|
|
Assert.NotNull(dequeuedFirst?.Event.WorkerTimestamp);
|
|
Assert.Equal(10, dequeuedFirst?.Event.ItemHandle);
|
|
Assert.Equal(11, dequeuedSecond?.Event.ItemHandle);
|
|
Assert.False(queue.TryDequeue(out _));
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies that Enqueue takes ownership of the passed event instead of
|
|
/// cloning it: the dequeued instance is the very reference passed in, and
|
|
/// the worker sequence/timestamp are stamped on that same instance
|
|
/// (WRK-11).
|
|
/// </summary>
|
|
[Fact]
|
|
public void Enqueue_TakesOwnershipOfPassedEventInstance()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 4);
|
|
MxEvent original = CreateEvent(MxEventFamily.OnDataChange, itemHandle: 10);
|
|
|
|
queue.Enqueue(original);
|
|
|
|
Assert.True(queue.TryDequeue(out WorkerEvent? dequeued));
|
|
Assert.Same(original, dequeued?.Event);
|
|
Assert.Equal(1UL, original.WorkerSequence);
|
|
Assert.NotNull(original.WorkerTimestamp);
|
|
}
|
|
|
|
/// <summary>Verifies that Drain removes at most the requested number of events.</summary>
|
|
[Fact]
|
|
public void Drain_RemovesAtMostRequestedEvents()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 4);
|
|
queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 10));
|
|
queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 11));
|
|
queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 12));
|
|
|
|
IReadOnlyList<WorkerEvent> drained = queue.Drain(maxEvents: 2);
|
|
|
|
Assert.Equal(2, drained.Count);
|
|
Assert.Equal(10, drained[0].Event.ItemHandle);
|
|
Assert.Equal(11, drained[1].Event.ItemHandle);
|
|
Assert.Equal(1, queue.Count);
|
|
}
|
|
|
|
/// <summary>Verifies that Drain with maxEvents 0 drains every queued event.</summary>
|
|
[Fact]
|
|
public void Drain_WithZeroMaxEvents_DrainsAllEvents()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 4);
|
|
queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 10));
|
|
queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 11));
|
|
queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 12));
|
|
|
|
IReadOnlyList<WorkerEvent> drained = queue.Drain(maxEvents: 0);
|
|
|
|
Assert.Equal(3, drained.Count);
|
|
Assert.Equal(new[] { 10, 11, 12 }, new[]
|
|
{
|
|
drained[0].Event.ItemHandle,
|
|
drained[1].Event.ItemHandle,
|
|
drained[2].Event.ItemHandle,
|
|
});
|
|
Assert.Equal(0, queue.Count);
|
|
}
|
|
|
|
/// <summary>Verifies that draining an empty queue returns an empty list.</summary>
|
|
[Fact]
|
|
public void Drain_WhenQueueIsEmpty_ReturnsEmptyList()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 4);
|
|
|
|
Assert.Empty(queue.Drain(maxEvents: 0));
|
|
Assert.Empty(queue.Drain(maxEvents: 5));
|
|
Assert.Equal(0, queue.Count);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies the byte-budgeted drain stops before the budget is exceeded, leaves the
|
|
/// remainder queued in order, and reports the exact remaining count (WRK-21). Events that
|
|
/// do not fit must never be dequeued — dequeuing them is how the pre-fix drain lost events
|
|
/// when the reply frame was rejected.
|
|
/// </summary>
|
|
[Fact]
|
|
public void Drain_ByteBudget_StopsBeforeBudgetAndLeavesRemainderQueued()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 8);
|
|
for (int itemHandle = 0; itemHandle < 5; itemHandle++)
|
|
{
|
|
queue.Enqueue(CreateEventWithPayload(itemHandle, payloadLength: 512));
|
|
}
|
|
|
|
int perEventCost = MeasureDrainCost(payloadLength: 512);
|
|
|
|
// Budget for exactly two events (plus a sliver too small for a third).
|
|
IReadOnlyList<WorkerEvent> drained =
|
|
queue.Drain(maxEvents: 0, maxTotalBytes: (perEventCost * 2) + (perEventCost / 2)).Events;
|
|
|
|
Assert.Equal(2, drained.Count);
|
|
Assert.Equal(0, drained[0].Event.ItemHandle);
|
|
Assert.Equal(1, drained[1].Event.ItemHandle);
|
|
Assert.Equal(3, queue.Count);
|
|
|
|
// The undrained remainder is still present, still in order.
|
|
IReadOnlyList<WorkerEvent> rest = queue.Drain(maxEvents: 0);
|
|
Assert.Equal(new[] { 2, 3, 4 }, new[] { rest[0].Event.ItemHandle, rest[1].Event.ItemHandle, rest[2].Event.ItemHandle });
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies the byte-budgeted drain reports truncation and the exact remaining count so the
|
|
/// DrainEvents reply can tell the caller to drain again.
|
|
/// </summary>
|
|
[Fact]
|
|
public void Drain_ByteBudget_ReportsTruncationAndRemainingCount()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 8);
|
|
for (int itemHandle = 0; itemHandle < 4; itemHandle++)
|
|
{
|
|
queue.Enqueue(CreateEventWithPayload(itemHandle, payloadLength: 256));
|
|
}
|
|
|
|
// One-and-a-half events' worth of budget: the head fits, the next does not, and the next is
|
|
// comfortably smaller than the whole budget so it is a plain truncation rather than the
|
|
// oversized-head case.
|
|
WorkerEventDrainResult result = queue.Drain(
|
|
maxEvents: 0,
|
|
maxTotalBytes: MeasureDrainCost(payloadLength: 256) * 3 / 2);
|
|
|
|
Assert.Single(result.Events);
|
|
Assert.True(result.TruncatedBySize);
|
|
Assert.Equal(3, result.RemainingCount);
|
|
Assert.Equal(0UL, result.OversizedHeadSequence);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies the degenerate case: a head event whose own serialized size exceeds the whole
|
|
/// budget is not drained (draining it would build an oversized reply or lose the event) and
|
|
/// its worker sequence is reported so an operator can find the offending tag.
|
|
/// </summary>
|
|
[Fact]
|
|
public void Drain_ByteBudget_OversizedHead_DrainsNothingAndReportsHeadSequence()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 8);
|
|
queue.Enqueue(CreateEventWithPayload(itemHandle: 0, payloadLength: 4096));
|
|
queue.Enqueue(CreateEventWithPayload(itemHandle: 1, payloadLength: 8));
|
|
|
|
WorkerEventDrainResult result = queue.Drain(maxEvents: 0, maxTotalBytes: 1024);
|
|
|
|
Assert.Empty(result.Events);
|
|
Assert.True(result.TruncatedBySize);
|
|
Assert.Equal(2, result.RemainingCount);
|
|
Assert.Equal(1UL, result.OversizedHeadSequence);
|
|
|
|
// The blocked event is still queued — it was never removed.
|
|
Assert.Equal(2, queue.Count);
|
|
}
|
|
|
|
/// <summary>
|
|
/// The no-loss half of the WRK-21 acceptance criterion, at full scale. Draining the review's
|
|
/// 10,000 byte-heavy events under a budget that fits only a fraction of them per call must
|
|
/// return every event exactly once and in order: the pre-fix drain removed events from the
|
|
/// queue before the reply was sized, so a rejected frame destroyed them. This runs at the
|
|
/// queue layer because the property is the queue's, and because the pipe harness that covers
|
|
/// the same walk end to end cannot sustain hundreds of large round trips.
|
|
/// </summary>
|
|
[Fact]
|
|
public void Drain_ByteBudget_RepeatedCalls_RecoverAllEventsInOrderWithoutLoss()
|
|
{
|
|
const int eventCount = 10_000;
|
|
const int payloadLength = 1_800;
|
|
MxAccessEventQueue queue = new(eventCount);
|
|
for (int index = 0; index < eventCount; index++)
|
|
{
|
|
queue.Enqueue(CreateEventWithPayload(index, payloadLength));
|
|
}
|
|
|
|
// A budget that fits roughly 35 events, so the walk takes hundreds of calls.
|
|
int budget = MeasureDrainCost(payloadLength) * 35;
|
|
List<ulong> recovered = new();
|
|
int calls = 0;
|
|
while (true)
|
|
{
|
|
WorkerEventDrainResult result = queue.Drain(maxEvents: 0, maxTotalBytes: budget);
|
|
calls++;
|
|
if (result.Events.Count == 0)
|
|
{
|
|
break;
|
|
}
|
|
|
|
foreach (WorkerEvent drained in result.Events)
|
|
{
|
|
recovered.Add(drained.Event.WorkerSequence);
|
|
}
|
|
|
|
Assert.Equal(eventCount - recovered.Count, result.RemainingCount);
|
|
Assert.True(calls < eventCount, "Drain made no progress.");
|
|
}
|
|
|
|
Assert.True(calls > 100, $"Expected the byte budget to split the drain, saw {calls} calls.");
|
|
Assert.Equal(eventCount, recovered.Count);
|
|
for (int index = 0; index < recovered.Count; index++)
|
|
{
|
|
Assert.Equal((ulong)(index + 1), recovered[index]);
|
|
}
|
|
|
|
Assert.Equal(0, queue.Count);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies the count cap still binds when the byte budget is generous: the byte cap is an
|
|
/// additional bound, not a replacement.
|
|
/// </summary>
|
|
[Fact]
|
|
public void Drain_ByteBudget_CountCapStillBinds()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 8);
|
|
for (int itemHandle = 0; itemHandle < 5; itemHandle++)
|
|
{
|
|
queue.Enqueue(CreateEventWithPayload(itemHandle, payloadLength: 16));
|
|
}
|
|
|
|
WorkerEventDrainResult result = queue.Drain(maxEvents: 2, maxTotalBytes: 1024 * 1024);
|
|
|
|
Assert.Equal(2, result.Events.Count);
|
|
Assert.False(result.TruncatedBySize);
|
|
Assert.Equal(3, result.RemainingCount);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Brackets the byte budget's per-event charge to the exact serialized size of the event.
|
|
/// The queue measures that size once at enqueue rather than during the drain, so this pins
|
|
/// the two ends of the charge against the very instance the queue holds: a budget one byte
|
|
/// short must refuse the head (and leave it queued, WRK-21), and a budget of exactly the
|
|
/// charge must ship it. An undercharge — sizing the event before Enqueue stamps its worker
|
|
/// sequence and timestamp, say — passes the first probe and breaks the frame guarantee; an
|
|
/// overcharge of even one byte fails the second. The <c>preStamped</c> case covers an event
|
|
/// that arrives with those two fields already filled, which Enqueue overwrites.
|
|
/// </summary>
|
|
/// <param name="preStamped">Whether the event carries a stale sequence/timestamp on arrival.</param>
|
|
[Theory]
|
|
[InlineData(false)]
|
|
[InlineData(true)]
|
|
public void Drain_ByteBudget_ChargesTheEventsExactSerializedSize(bool preStamped)
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 4);
|
|
MxEvent mxEvent = CreateEventWithPayload(itemHandle: 7, payloadLength: 300);
|
|
if (preStamped)
|
|
{
|
|
mxEvent.WorkerSequence = ulong.MaxValue;
|
|
mxEvent.WorkerTimestamp = Timestamp.FromDateTime(
|
|
new DateTime(1990, 1, 1, 0, 0, 0, DateTimeKind.Utc));
|
|
}
|
|
|
|
queue.Enqueue(mxEvent);
|
|
|
|
// Enqueue takes ownership but never mutates the event afterwards, so the retained reference
|
|
// measures exactly what the queue holds — no second drain is needed to learn the cost.
|
|
Assert.Equal(1UL, mxEvent.WorkerSequence);
|
|
int exactCost = DrainCostOf(new WorkerEvent { Event = mxEvent });
|
|
|
|
WorkerEventDrainResult refused = queue.Drain(maxEvents: 0, maxTotalBytes: exactCost - 1);
|
|
Assert.Empty(refused.Events);
|
|
Assert.True(refused.TruncatedBySize);
|
|
Assert.Equal(1UL, refused.OversizedHeadSequence);
|
|
Assert.Equal(1, queue.Count);
|
|
|
|
WorkerEventDrainResult drained = queue.Drain(maxEvents: 0, maxTotalBytes: exactCost);
|
|
Assert.Single(drained.Events);
|
|
Assert.Same(mxEvent, drained.Events[0].Event);
|
|
Assert.False(drained.TruncatedBySize);
|
|
Assert.Equal(0, queue.Count);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Pins the batch boundaries of a byte-budgeted walk over a mixed-size queue. Each batch is
|
|
/// checked against the true serialized sizes of the events it returned: no batch may exceed
|
|
/// the budget (an undercharge would build a reply past the frame maximum) and no batch may
|
|
/// stop early — the next event, at its real size, must not have fit (an overcharge would
|
|
/// ship frames smaller than the negotiated maximum allows). Both bounds come from the
|
|
/// drained events themselves, so any drift between the size memoized at enqueue and the real
|
|
/// one moves a boundary and fails here.
|
|
/// </summary>
|
|
[Fact]
|
|
public void Drain_ByteBudget_MixedSizes_KeepsBatchBoundariesOnTheRealSizes()
|
|
{
|
|
int[] payloadLengths = { 8, 512, 40, 2_048, 96, 1_200, 16, 700 };
|
|
const int eventCount = 240;
|
|
|
|
// Comfortably above the largest single event's cost, so the walk is never blocked by an
|
|
// oversized head and every stop is a genuine budget boundary.
|
|
const int budget = 4096;
|
|
|
|
MxAccessEventQueue queue = new(eventCount);
|
|
for (int index = 0; index < eventCount; index++)
|
|
{
|
|
queue.Enqueue(CreateEventWithPayload(index, payloadLengths[index % payloadLengths.Length]));
|
|
}
|
|
|
|
List<IReadOnlyList<WorkerEvent>> batches = new();
|
|
while (true)
|
|
{
|
|
WorkerEventDrainResult result = queue.Drain(maxEvents: 0, maxTotalBytes: budget);
|
|
if (result.Events.Count == 0)
|
|
{
|
|
Assert.Equal(0UL, result.OversizedHeadSequence);
|
|
break;
|
|
}
|
|
|
|
batches.Add(result.Events);
|
|
Assert.True(batches.Count <= eventCount, "Drain made no progress.");
|
|
}
|
|
|
|
Assert.Equal(0, queue.Count);
|
|
Assert.True(
|
|
batches.Count > 5,
|
|
$"Expected the byte budget to split the drain, saw {batches.Count} batches.");
|
|
|
|
ulong expectedSequence = 0;
|
|
for (int batchIndex = 0; batchIndex < batches.Count; batchIndex++)
|
|
{
|
|
IReadOnlyList<WorkerEvent> batch = batches[batchIndex];
|
|
int charged = 0;
|
|
foreach (WorkerEvent drained in batch)
|
|
{
|
|
charged += DrainCostOf(drained);
|
|
Assert.Equal(++expectedSequence, drained.Event.WorkerSequence);
|
|
}
|
|
|
|
Assert.True(
|
|
charged <= budget,
|
|
$"Batch {batchIndex} shipped {charged} bytes against a {budget} byte budget.");
|
|
|
|
if (batchIndex + 1 < batches.Count)
|
|
{
|
|
int nextCost = DrainCostOf(batches[batchIndex + 1][0]);
|
|
Assert.True(
|
|
charged + nextCost > budget,
|
|
$"Batch {batchIndex} stopped at {charged} bytes although the next event's {nextCost} still fit the {budget} byte budget.");
|
|
}
|
|
}
|
|
|
|
Assert.Equal((ulong)eventCount, expectedSequence);
|
|
}
|
|
|
|
/// <summary>
|
|
/// The other half of the WRK-21 head guarantee: a head refused for its size is drained
|
|
/// unchanged by a later call whose budget fits it. The size the queue charges lives with the
|
|
/// event across calls, so a refused attempt must neither consume nor alter it — and the
|
|
/// refusal itself is justified by the event's real serialized size.
|
|
/// </summary>
|
|
[Fact]
|
|
public void Drain_ByteBudget_RefusedHead_IsDrainedUnchangedOnceTheBudgetFitsIt()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 8);
|
|
queue.Enqueue(CreateEventWithPayload(itemHandle: 0, payloadLength: 4096));
|
|
queue.Enqueue(CreateEventWithPayload(itemHandle: 1, payloadLength: 8));
|
|
|
|
WorkerEventDrainResult refused = queue.Drain(maxEvents: 0, maxTotalBytes: 1024);
|
|
Assert.Empty(refused.Events);
|
|
Assert.True(refused.TruncatedBySize);
|
|
Assert.Equal(1UL, refused.OversizedHeadSequence);
|
|
Assert.Equal(2, queue.Count);
|
|
|
|
WorkerEventDrainResult retried = queue.Drain(maxEvents: 0, maxTotalBytes: 64 * 1024);
|
|
|
|
Assert.Equal(2, retried.Events.Count);
|
|
Assert.Equal(1UL, retried.Events[0].Event.WorkerSequence);
|
|
Assert.Equal(2UL, retried.Events[1].Event.WorkerSequence);
|
|
Assert.False(retried.TruncatedBySize);
|
|
Assert.Equal(0, retried.RemainingCount);
|
|
Assert.Equal(0, queue.Count);
|
|
|
|
Assert.True(
|
|
DrainCostOf(retried.Events[0]) > 1024,
|
|
"The head was refused although its real cost fits the 1024-byte budget it was refused under.");
|
|
Assert.True(
|
|
DrainCostOf(retried.Events[0]) + DrainCostOf(retried.Events[1]) <= 64 * 1024,
|
|
"The retried batch exceeded the budget it was drained under.");
|
|
}
|
|
|
|
/// <summary>Verifies that Enqueue is rejected after a fault is recorded manually.</summary>
|
|
[Fact]
|
|
public void Enqueue_AfterRecordFault_ThrowsInvalidOperationException()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 4);
|
|
queue.RecordFault(new WorkerFault
|
|
{
|
|
Category = WorkerFaultCategory.MxaccessEventConversionFailed,
|
|
});
|
|
|
|
Assert.Throws<InvalidOperationException>(
|
|
() => queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 10)));
|
|
Assert.Equal(0, queue.Count);
|
|
}
|
|
|
|
/// <summary>Verifies that Enqueue records an overflow fault and rejects new events when capacity is exceeded.</summary>
|
|
[Fact]
|
|
public void Enqueue_WhenCapacityIsExceeded_RecordsOverflowFaultAndRejectsNewEvents()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 1);
|
|
queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 10));
|
|
|
|
MxAccessEventQueueOverflowException overflow = Assert.Throws<MxAccessEventQueueOverflowException>(
|
|
() => queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 11)));
|
|
|
|
Assert.Equal(1, overflow.Capacity);
|
|
Assert.True(queue.IsFaulted);
|
|
Assert.Equal(WorkerFaultCategory.QueueOverflow, queue.Fault?.Category);
|
|
Assert.Equal(ProtocolStatusCode.WorkerUnavailable, queue.Fault?.ProtocolStatus.Code);
|
|
Assert.Throws<InvalidOperationException>(
|
|
() => queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 12)));
|
|
}
|
|
|
|
/// <summary>Verifies that RecordFault keeps the first recorded fault.</summary>
|
|
[Fact]
|
|
public void RecordFault_KeepsFirstFault()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 1);
|
|
queue.RecordFault(new WorkerFault
|
|
{
|
|
Category = WorkerFaultCategory.MxaccessEventConversionFailed,
|
|
});
|
|
queue.RecordFault(new WorkerFault
|
|
{
|
|
Category = WorkerFaultCategory.QueueOverflow,
|
|
});
|
|
|
|
Assert.True(queue.IsFaulted);
|
|
Assert.Equal(WorkerFaultCategory.MxaccessEventConversionFailed, queue.Fault?.Category);
|
|
}
|
|
|
|
// Wake-signal timings. The fallback is far longer than the patience deliberately: every wait
|
|
// below asserts "the signal completed this", which is only a faithful claim while the fallback
|
|
// timeout cannot have completed it within the patience window. The patience itself is generous
|
|
// so a loaded CI box cannot fail a test that is not about latency.
|
|
private static readonly TimeSpan WakeFallback = TimeSpan.FromSeconds(30);
|
|
private static readonly TimeSpan WakePatience = TimeSpan.FromSeconds(5);
|
|
|
|
/// <summary>
|
|
/// Verifies the queue wakes a parked waiter as soon as an event is enqueued, rather than
|
|
/// leaving it to time out. This is what removes the drain loop's latency floor: before the
|
|
/// signal existed, an event arriving at an idle queue waited out the loop's whole poll tick.
|
|
/// </summary>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
[Fact]
|
|
public async Task WaitForEventsAsync_EnqueueAfterIdle_CompletesWithoutWaitingTheFallback()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 4);
|
|
|
|
Task wait = queue.WaitForEventsAsync(WakeFallback, CancellationToken.None);
|
|
Assert.False(wait.IsCompleted);
|
|
|
|
queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 10));
|
|
|
|
Assert.Same(wait, await Task.WhenAny(wait, Task.Delay(WakePatience)));
|
|
await wait;
|
|
Assert.Equal(1, queue.Count);
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies a fault recorded while the waiter is parked wakes it too. The drain loop
|
|
/// discovers faults by calling <c>DrainFault()</c> at the top of each pass, so without this
|
|
/// signal an overflow or conversion fault would not be reported until the loop's fallback
|
|
/// tick expired.
|
|
/// </summary>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
[Fact]
|
|
public async Task WaitForEventsAsync_RecordFaultWhileParked_WakesTheWaiter()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 4);
|
|
|
|
Task wait = queue.WaitForEventsAsync(WakeFallback, CancellationToken.None);
|
|
Assert.False(wait.IsCompleted);
|
|
|
|
queue.RecordFault(new WorkerFault
|
|
{
|
|
Category = WorkerFaultCategory.MxaccessEventConversionFailed,
|
|
});
|
|
|
|
Assert.Same(wait, await Task.WhenAny(wait, Task.Delay(WakePatience)));
|
|
await wait;
|
|
Assert.NotNull(queue.DrainFault());
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies the one-permit cap loses no wakeups. A burst that lands while nobody is waiting
|
|
/// leaves exactly one pending wake — the waiter that consumes it drains the whole burst, so
|
|
/// coalescing costs nothing — and that consumed wake is not replayed: the next wait parks
|
|
/// until a new enqueue signals it.
|
|
/// </summary>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
[Fact]
|
|
public async Task WaitForEventsAsync_BurstWhileNoWaiter_CoalescesToOneWakeThatLosesNothing()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 16);
|
|
for (int itemHandle = 0; itemHandle < 5; itemHandle++)
|
|
{
|
|
queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle));
|
|
}
|
|
|
|
Task firstWait = queue.WaitForEventsAsync(WakeFallback, CancellationToken.None);
|
|
Assert.Same(firstWait, await Task.WhenAny(firstWait, Task.Delay(WakePatience)));
|
|
await firstWait;
|
|
|
|
// One wake, the whole burst: the waiter re-drains everything queued, which is why capping
|
|
// the signal at a single permit cannot drop an event.
|
|
Assert.Equal(5, queue.Drain(maxEvents: 0).Count);
|
|
|
|
Task secondWait = queue.WaitForEventsAsync(WakeFallback, CancellationToken.None);
|
|
Assert.False(secondWait.IsCompleted);
|
|
|
|
queue.Enqueue(CreateEvent(MxEventFamily.OnDataChange, itemHandle: 99));
|
|
|
|
Assert.Same(secondWait, await Task.WhenAny(secondWait, Task.Delay(WakePatience)));
|
|
await secondWait;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Verifies the timeout still bounds an unsignalled wait, from both sides: the wait ends
|
|
/// without a signal (the fallback survives as a ceiling, so a state change reached by some
|
|
/// future path that does not signal is still observed on the next pass rather than never)
|
|
/// and it does not end early (the wait really is the timeout, not an already-armed permit
|
|
/// completing it instantly). A longer-than-production timeout is used so the lower bound
|
|
/// carries a wide margin over timer resolution.
|
|
/// </summary>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
[Fact]
|
|
public async Task WaitForEventsAsync_WithNoSignal_CompletesAtTheFallbackTimeout()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 4);
|
|
TimeSpan fallback = TimeSpan.FromMilliseconds(200);
|
|
Stopwatch elapsed = Stopwatch.StartNew();
|
|
|
|
Task wait = queue.WaitForEventsAsync(fallback, CancellationToken.None);
|
|
|
|
Assert.Same(wait, await Task.WhenAny(wait, Task.Delay(WakePatience)));
|
|
await wait;
|
|
elapsed.Stop();
|
|
|
|
// Half the timeout: far enough below it to be immune to timer resolution, far enough above
|
|
// zero to fail an implementation that returned a completed task instead of waiting.
|
|
Assert.True(
|
|
elapsed.Elapsed >= TimeSpan.FromMilliseconds(100),
|
|
$"Unsignalled wait returned after {elapsed.ElapsedMilliseconds} ms, well inside its {fallback.TotalMilliseconds} ms fallback.");
|
|
}
|
|
|
|
/// <summary>Verifies a cancelled wait unwinds instead of hanging until the fallback expires.</summary>
|
|
/// <returns>A task that represents the asynchronous operation.</returns>
|
|
[Fact]
|
|
public async Task WaitForEventsAsync_WhenCancelled_Throws()
|
|
{
|
|
MxAccessEventQueue queue = new(capacity: 4);
|
|
using CancellationTokenSource cancellation = new();
|
|
|
|
Task wait = queue.WaitForEventsAsync(WakeFallback, cancellation.Token);
|
|
cancellation.Cancel();
|
|
|
|
await Assert.ThrowsAnyAsync<OperationCanceledException>(async () => await wait);
|
|
}
|
|
|
|
/// <summary>
|
|
/// The queue capacity comes from the launcher-set environment variable;
|
|
/// a missing, unparseable, below-floor, or above-ceiling value must fall
|
|
/// back to the 10,000 default rather than throw. A bad environment value
|
|
/// must never stop the worker's session from starting.
|
|
/// </summary>
|
|
/// <param name="environmentValue">Raw environment value under test.</param>
|
|
/// <param name="expected">Expected resolved capacity.</param>
|
|
[Theory]
|
|
[InlineData(null, MxAccessEventQueue.DefaultCapacity)]
|
|
[InlineData("", MxAccessEventQueue.DefaultCapacity)]
|
|
[InlineData("not-a-number", MxAccessEventQueue.DefaultCapacity)]
|
|
[InlineData("0", MxAccessEventQueue.DefaultCapacity)]
|
|
[InlineData("-5", MxAccessEventQueue.DefaultCapacity)]
|
|
[InlineData("999", MxAccessEventQueue.DefaultCapacity)]
|
|
[InlineData("1000001", MxAccessEventQueue.DefaultCapacity)]
|
|
[InlineData("1000", 1000)]
|
|
[InlineData("1000000", 1000000)]
|
|
[InlineData("50000", 50000)]
|
|
public void ResolveCapacity_WithEnvironmentValue_FallsBackToDefaultWhenUnusable(
|
|
string? environmentValue,
|
|
int expected)
|
|
{
|
|
string? original = Environment.GetEnvironmentVariable(
|
|
MxAccessEventQueue.CapacityEnvironmentVariableName);
|
|
try
|
|
{
|
|
Environment.SetEnvironmentVariable(
|
|
MxAccessEventQueue.CapacityEnvironmentVariableName,
|
|
environmentValue);
|
|
|
|
Assert.Equal(expected, MxAccessEventQueue.ResolveCapacity());
|
|
}
|
|
finally
|
|
{
|
|
Environment.SetEnvironmentVariable(
|
|
MxAccessEventQueue.CapacityEnvironmentVariableName,
|
|
original);
|
|
}
|
|
}
|
|
|
|
/// <summary>
|
|
/// A resolved capacity is what the queue is actually built with, so the
|
|
/// configured headroom reaches the overflow check rather than only the
|
|
/// resolver.
|
|
/// </summary>
|
|
[Fact]
|
|
public void Constructor_WithResolvedCapacity_UsesEnvironmentValue()
|
|
{
|
|
string? original = Environment.GetEnvironmentVariable(
|
|
MxAccessEventQueue.CapacityEnvironmentVariableName);
|
|
try
|
|
{
|
|
Environment.SetEnvironmentVariable(
|
|
MxAccessEventQueue.CapacityEnvironmentVariableName,
|
|
"2500");
|
|
|
|
MxAccessEventQueue queue = new(MxAccessEventQueue.ResolveCapacity());
|
|
|
|
Assert.Equal(2500, queue.Capacity);
|
|
}
|
|
finally
|
|
{
|
|
Environment.SetEnvironmentVariable(
|
|
MxAccessEventQueue.CapacityEnvironmentVariableName,
|
|
original);
|
|
}
|
|
}
|
|
|
|
// Mirrors MxAccessEventQueue's per-event repeated-field allowance. Kept local rather than made
|
|
// public on the queue: the byte-budget tests state their budgets in units of that charge, so a
|
|
// change to it should surface here as a failing bound instead of silently moving with the code.
|
|
private const int RepeatedFieldOverheadBytes = 8;
|
|
|
|
/// <summary>
|
|
/// Measures what the queue charges one event of the given payload size against the byte budget:
|
|
/// the serialized <see cref="WorkerEvent"/> as it exists after Enqueue (sequence and timestamp
|
|
/// stamped) plus the repeated-field allowance. The probe uses item handle 0, a proto3 default
|
|
/// that is not serialized, so this is a lower bound on the fixtures' real per-event cost — the
|
|
/// budgets above carry slack rather than assuming byte equality.
|
|
/// </summary>
|
|
/// <param name="payloadLength">Length of the event's raw-status payload string.</param>
|
|
/// <returns>The per-event byte cost.</returns>
|
|
/// <summary>
|
|
/// What the byte budget must charge for an already-stamped event: its true serialized size plus
|
|
/// the repeated-field allowance. Measured from the generated <c>CalculateSize()</c> so the
|
|
/// budget tests bound the queue's memoized size against the real one rather than against a
|
|
/// second copy of the queue's own arithmetic.
|
|
/// </summary>
|
|
/// <param name="workerEvent">Event as the queue holds it, sequence and timestamp stamped.</param>
|
|
/// <returns>The per-event byte cost.</returns>
|
|
private static int DrainCostOf(WorkerEvent workerEvent)
|
|
{
|
|
return workerEvent.CalculateSize() + RepeatedFieldOverheadBytes;
|
|
}
|
|
|
|
private static int MeasureDrainCost(int payloadLength)
|
|
{
|
|
MxAccessEventQueue probe = new(capacity: 1);
|
|
probe.Enqueue(CreateEventWithPayload(0, payloadLength));
|
|
Assert.True(probe.TryDequeue(out WorkerEvent? probeEvent));
|
|
return probeEvent!.CalculateSize() + RepeatedFieldOverheadBytes;
|
|
}
|
|
|
|
/// <summary>
|
|
/// Builds a byte-heavy event: a large string field is the cheapest stand-in for the array/string
|
|
/// <see cref="MxValue"/> payloads that make a count-capped drain overshoot the frame maximum.
|
|
/// </summary>
|
|
/// <param name="itemHandle">Item handle identifying the event in assertions.</param>
|
|
/// <param name="payloadLength">Length of the raw-status payload string.</param>
|
|
/// <returns>The constructed event.</returns>
|
|
private static MxEvent CreateEventWithPayload(int itemHandle, int payloadLength)
|
|
{
|
|
MxEvent mxEvent = CreateEvent(MxEventFamily.OnDataChange, itemHandle);
|
|
mxEvent.RawStatus = new string('x', payloadLength);
|
|
return mxEvent;
|
|
}
|
|
|
|
private static MxEvent CreateEvent(
|
|
MxEventFamily family,
|
|
int itemHandle)
|
|
{
|
|
MxEvent mxEvent = new()
|
|
{
|
|
Family = family,
|
|
SessionId = "session-1",
|
|
ServerHandle = 1,
|
|
ItemHandle = itemHandle,
|
|
};
|
|
|
|
switch (family)
|
|
{
|
|
case MxEventFamily.OnWriteComplete:
|
|
mxEvent.OnWriteComplete = new OnWriteCompleteEvent();
|
|
break;
|
|
|
|
default:
|
|
mxEvent.OnDataChange = new OnDataChangeEvent();
|
|
break;
|
|
}
|
|
|
|
return mxEvent;
|
|
}
|
|
}
|