From cf1b3d40a5a6f057b2aac39f07f84647618ab536 Mon Sep 17 00:00:00 2001 From: Joseph Doherty Date: Thu, 9 Jul 2026 15:17:11 -0400 Subject: [PATCH] perf(gateway): replay ring LinkedList -> circular array (GWC-14) Replace the replay buffer's LinkedList (a node allocation per retained event on the fan-out hot path) with a preallocated ReplayEntry[] ring sized to ReplayBufferCapacity, tracked by _replayHead (oldest) + _replayCount. Appending a retained event now allocates nothing. Behavior preserved exactly: ascending-WorkerSequence append order; capacity eviction (overwrite head + advance when full); time trim via _timeProvider.GetUtcNow() cutoff at the same three call sites; oldest-read for ReplayGap math; both replay-from-sequence query paths + highest-seen tracking; same _replayLock. Capacity-0 stays retain-nothing (guarded early return, no modulo-by-zero). Server build clean (0 warnings); Distributor/Replay tests 33/33 incl. two new cases (multi-wrap ring keeps newest in order; capacity-1 overwrite + gap). Claude-Session: https://claude.ai/code/session_01DMXXvNuPekkkrTEyPNxEkW --- .../Sessions/SessionEventDistributor.cs | 73 ++++++++++------ .../Sessions/SessionEventDistributorTests.cs | 84 +++++++++++++++++++ 2 files changed, 132 insertions(+), 25 deletions(-) diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs index 8d94da3..29e910a 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs @@ -98,14 +98,21 @@ public sealed class SessionEventDistributor : IAsyncDisposable private readonly object _lifecycleLock = new(); // Replay ring buffer. Appended on the pump thread and queried from arbitrary - // threads via TryGetReplayFrom, so every access is under _replayLock. The deque - // keeps events in ascending WorkerSequence order (the pump fans in source order), - // so the oldest retained event is always at the front. Capacity == 0 disables - // retention; RetentionSeconds <= 0 disables age-based eviction. + // threads via TryGetReplayFrom, so every access is under _replayLock. Backed by a + // fixed-size circular array preallocated to the capacity so appending a retained + // event allocates no node (the LinkedList this replaced allocated one node per + // event on the fan-out hot path). Events are kept in ascending WorkerSequence order + // (the pump fans in source order): _replayHead is the logical front (oldest retained + // event) and _replayCount entries follow it, wrapping modulo the array length. The + // logical entry at position i is _replayBuffer[(_replayHead + i) % Length]. Capacity + // == 0 disables retention (the array is empty and never indexed); RetentionSeconds + // <= 0 disables age-based eviction. private readonly int _replayBufferCapacity; private readonly TimeSpan _replayRetention; private readonly bool _ageEvictionEnabled; - private readonly LinkedList _replayBuffer = new(); + private readonly ReplayEntry[] _replayBuffer; + private int _replayHead; + private int _replayCount; private readonly object _replayLock = new(); private bool _anyEventSeen; private ulong _highestSequenceSeen; @@ -245,6 +252,7 @@ public sealed class SessionEventDistributor : IAsyncDisposable _overflowHandler = overflowHandler; _shutdownTimeout = DefaultShutdownTimeout; _replayBufferCapacity = replayBufferCapacity; + _replayBuffer = new ReplayEntry[replayBufferCapacity]; _ageEvictionEnabled = replayRetentionSeconds > 0; _replayRetention = _ageEvictionEnabled ? TimeSpan.FromSeconds(replayRetentionSeconds) @@ -452,24 +460,25 @@ public sealed class SessionEventDistributor : IAsyncDisposable List newer = []; ulong highestReplayed = afterSequence; - if (_replayBuffer.Count == 0) + if (_replayCount == 0) { gap = _anyEventSeen && afterSequence < _highestSequenceSeen; oldestAvailableSequence = 0; // meaningful only when gap == true; 0 here since nothing is retained } else { - ulong oldestRetained = _replayBuffer.First!.Value.Event.WorkerSequence; + ulong oldestRetained = ReplayEntryAt(0).Event.WorkerSequence; gap = oldestRetained > 0 && afterSequence < oldestRetained - 1; // Per the contract on OldestAvailableSequence: meaningful only when gap == true. oldestAvailableSequence = gap ? oldestRetained : 0; - foreach (ReplayEntry entry in _replayBuffer) + for (int i = 0; i < _replayCount; i++) { - if (entry.Event.WorkerSequence > afterSequence) + MxEvent retained = ReplayEntryAt(i).Event; + if (retained.WorkerSequence > afterSequence) { - newer.Add(entry.Event); - highestReplayed = entry.Event.WorkerSequence; + newer.Add(retained); + highestReplayed = retained.WorkerSequence; } } } @@ -729,7 +738,7 @@ public sealed class SessionEventDistributor : IAsyncDisposable { EvictAged(); - if (_replayBuffer.Count == 0) + if (_replayCount == 0) { events = []; // Nothing retained. The caller missed events only if it is behind the @@ -738,7 +747,7 @@ public sealed class SessionEventDistributor : IAsyncDisposable return true; } - ulong oldestRetained = _replayBuffer.First!.Value.Event.WorkerSequence; + ulong oldestRetained = ReplayEntryAt(0).Event.WorkerSequence; // A gap exists when at least one event newer than afterSequence was evicted, // i.e. afterSequence sits below the oldest-retained-minus-one boundary. @@ -750,11 +759,12 @@ public sealed class SessionEventDistributor : IAsyncDisposable // O(n) scan over the retained buffer — acceptable because TryGetReplayFrom // is only called on subscriber reconnect, never on the hot fan-out path. List newer = []; - foreach (ReplayEntry entry in _replayBuffer) + for (int i = 0; i < _replayCount; i++) { - if (entry.Event.WorkerSequence > afterSequence) + MxEvent retained = ReplayEntryAt(i).Event; + if (retained.WorkerSequence > afterSequence) { - newer.Add(entry.Event); + newer.Add(retained); } } @@ -780,30 +790,43 @@ public sealed class SessionEventDistributor : IAsyncDisposable return; } - _replayBuffer.AddLast(new ReplayEntry(mxEvent, _timeProvider.GetUtcNow())); - - // Capacity eviction: drop oldest until within bound. - while (_replayBuffer.Count > _replayBufferCapacity) + // Append at the logical tail. When the ring is full the oldest entry is + // overwritten in place (its slot becomes the new tail) and the head advances, + // so the newest _replayBufferCapacity events are retained with no allocation. + ReplayEntry entry = new(mxEvent, _timeProvider.GetUtcNow()); + if (_replayCount < _replayBufferCapacity) { - _replayBuffer.RemoveFirst(); + _replayBuffer[(_replayHead + _replayCount) % _replayBufferCapacity] = entry; + _replayCount++; + } + else + { + _replayBuffer[_replayHead] = entry; + _replayHead = (_replayHead + 1) % _replayBufferCapacity; } EvictAged(); } } - // Must be called under _replayLock. Drops entries older than the retention window. + // Returns the logical entry at position i (0 == oldest retained). Must be called + // under _replayLock with 0 <= i < _replayCount (so the array length is nonzero). + private ReplayEntry ReplayEntryAt(int i) => _replayBuffer[(_replayHead + i) % _replayBuffer.Length]; + + // Must be called under _replayLock. Drops entries older than the retention window + // by advancing the head past them (no dealloc; slots are reused on the next append). private void EvictAged() { - if (!_ageEvictionEnabled || _replayBuffer.Count == 0) + if (!_ageEvictionEnabled || _replayCount == 0) { return; } DateTimeOffset cutoff = _timeProvider.GetUtcNow() - _replayRetention; - while (_replayBuffer.First is { } first && first.Value.RetainedAt < cutoff) + while (_replayCount > 0 && _replayBuffer[_replayHead].RetainedAt < cutoff) { - _replayBuffer.RemoveFirst(); + _replayHead = (_replayHead + 1) % _replayBuffer.Length; + _replayCount--; } } diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionEventDistributorTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionEventDistributorTests.cs index 1bff529..75a2544 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionEventDistributorTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Sessions/SessionEventDistributorTests.cs @@ -185,6 +185,90 @@ public sealed class SessionEventDistributorTests Assert.Equal(new ulong[] { 3, 4, 5 }, replay.Select(e => e.WorkerSequence)); } + /// + /// Appending far more events than the capacity wraps the circular buffer's head + /// index past the array boundary multiple times; the retained window stays the + /// newest capacity events, in ascending order, and a replay from before the + /// window reports a gap and returns the whole retained tail. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReplayBuffer_WrapsRingMultipleTimes_RetainsNewestInAscendingOrder() + { + Channel source = Channel.CreateUnbounded(); + await using SessionEventDistributor distributor = CreateDistributor( + source.Reader, + replayBufferCapacity: 4, + replayRetentionSeconds: 0); + await distributor.StartAsync(CancellationToken.None); + + // 13 events through a capacity-4 ring advances the head index 13 - 4 = 9 slots, + // wrapping the 4-slot array's boundary more than twice. + using IEventSubscriberLease lease = distributor.Register(); + for (ulong sequence = 1; sequence <= 13; sequence++) + { + source.Writer.TryWrite(Event(sequence)); + } + + for (ulong sequence = 1; sequence <= 13; sequence++) + { + MxEvent e = await ReadOneAsync(lease.Reader); + Assert.Equal(sequence, e.WorkerSequence); + } + + // Newest four (10, 11, 12, 13) retained in ascending order despite the wraps. + bool found = distributor.TryGetReplayFrom(0, out IReadOnlyList replay, out bool gap); + + Assert.True(found); + Assert.True(gap); + Assert.Equal(new ulong[] { 10, 11, 12, 13 }, replay.Select(e => e.WorkerSequence)); + + // A replay from inside the wrapped window returns only the newer entries, no gap, + // proving the modular scan reads the logical order and not the physical slots. + bool foundInner = distributor.TryGetReplayFrom(11, out IReadOnlyList inner, out bool innerGap); + + Assert.True(foundInner); + Assert.False(innerGap); + Assert.Equal(new ulong[] { 12, 13 }, inner.Select(e => e.WorkerSequence)); + } + + /// + /// A capacity-1 ring retains only the single newest event; each append overwrites + /// the sole slot, and a replay from before it reports a gap. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReplayBuffer_Capacity1_RetainsOnlyNewest() + { + Channel source = Channel.CreateUnbounded(); + await using SessionEventDistributor distributor = CreateDistributor( + source.Reader, + replayBufferCapacity: 1, + replayRetentionSeconds: 0); + await distributor.StartAsync(CancellationToken.None); + + using IEventSubscriberLease lease = distributor.Register(); + for (ulong sequence = 1; sequence <= 3; sequence++) + { + source.Writer.TryWrite(Event(sequence)); + _ = await ReadOneAsync(lease.Reader); + } + + // Only sequence 3 is retained; a request from 0 missed 1 and 2 => gap. + bool found = distributor.TryGetReplayFrom(0, out IReadOnlyList replay, out bool gap); + + Assert.True(found); + Assert.True(gap); + Assert.Equal(new ulong[] { 3 }, replay.Select(e => e.WorkerSequence)); + + // A request from exactly the newest retained sequence is caught up: empty, no gap. + bool foundCaughtUp = distributor.TryGetReplayFrom(3, out IReadOnlyList caughtUp, out bool caughtUpGap); + + Assert.True(foundCaughtUp); + Assert.False(caughtUpGap); + Assert.Empty(caughtUp); + } + /// Requesting replay from a sequence still inside the retained window returns only the newer events, with no gap. /// A task that represents the asynchronous operation. [Fact]