refactor(sessions): _subscribers to plain Dictionary — every access is under _lifecycleLock

This commit is contained in:
Joseph Doherty
2026-08-15 20:07:15 -04:00
parent 94dbe9f1af
commit 59420d8568
@@ -1,4 +1,3 @@
using System.Collections.Concurrent;
using System.Threading.Channels;
using ZB.MOM.WW.MxGateway.Contracts.Proto;
@@ -66,18 +65,20 @@ public delegate void SubscriberOverflowHandler(bool isOnlySubscriber, bool isInt
/// <c>EventStreamService.ProduceEventsAsync</c> ordering.
/// </para>
/// <para>
/// <b>Concurrency.</b> The subscriber set is a
/// <see cref="ConcurrentDictionary{TKey, TValue}"/> keyed by a monotonic id, used
/// for keyed add/remove only. Fan-out does NOT enumerate the dictionary: every
/// mutation (<see cref="Register"/>, <see cref="RegisterWithReplay"/>, lease
/// disposal, overflow disconnect) happens inside the <c>_lifecycleLock</c> critical
/// section and rebuilds an immutable copy-on-write <c>Subscriber[]</c> snapshot,
/// which the pump reads once per event. This matters because
/// <c>ConcurrentDictionary.Values</c> is a PROPERTY that acquires every internal
/// lock and materializes a fresh <c>List</c> plus a read-only wrapper on each call
/// — per event, on the hot fan-out path. The subscriber set is tiny (one to a
/// handful) and mutates rarely, so paying a full array rebuild per registration to
/// make fan-out a bare array walk is the right trade. No lock is held across an
/// <b>Concurrency.</b> The subscriber set is a plain
/// <see cref="Dictionary{TKey, TValue}"/> keyed by a monotonic id, used for keyed
/// add/remove only. It needs no concurrent collection type because it is never
/// touched outside the <c>_lifecycleLock</c> critical section: every mutation
/// (<see cref="Register"/>, <see cref="RegisterWithReplay"/>, lease disposal,
/// overflow disconnect) and every read (the terminal completion sweep, the snapshot
/// rebuild) holds that lock, and each mutation rebuilds an immutable copy-on-write
/// <c>Subscriber[]</c> snapshot inside the same section. The lock-free readers see only that snapshot,
/// never the dictionary: the pump reads it once per event and
/// <see cref="SubscriberCount"/> reads its length. Fan-out therefore does NOT
/// enumerate the dictionary — it walks a captured array, with no dictionary
/// traversal and no per-event allocation on the hot path. The subscriber set is
/// tiny (one to a handful) and mutates rarely, so paying a full array rebuild per
/// registration to buy that is the right trade. No lock is held across an
/// <c>await</c>. Each subscriber channel has a single writer — the pump — so
/// per-channel writes never race. A subscriber registered after the pump captured
/// the array for the in-flight event misses that event, which matches "late
@@ -104,7 +105,11 @@ public sealed class SessionEventDistributor : IAsyncDisposable
private readonly TimeSpan _shutdownTimeout;
private readonly ILogger<SessionEventDistributor> _logger;
private readonly TimeProvider _timeProvider;
private readonly ConcurrentDictionary<long, Subscriber> _subscribers = new();
// Keyed subscriber set. Touched ONLY under _lifecycleLock (add in RegisterSubscriber and
// RegisterWithReplay, remove in RemoveSubscriber, read in CompleteAllSubscribers and
// RebuildSubscriberSnapshot), which is why a plain Dictionary suffices: lock-free readers
// never see this field, they read _subscriberSnapshot below.
private readonly Dictionary<long, Subscriber> _subscribers = [];
private readonly CancellationTokenSource _shutdownCts = new();
private readonly object _lifecycleLock = new();
@@ -120,7 +125,7 @@ public sealed class SessionEventDistributor : IAsyncDisposable
// may legitimately observe the previous array, which IS the documented "late subscribers
// see events after they register" window. Where visibility must be guaranteed — the
// RegisterWithReplay handoff — it comes from the _replayLock edge, not from Volatile.
// See the type remarks for why fan-out must not touch ConcurrentDictionary.Values.
// See the type remarks for why fan-out walks this array instead of enumerating _subscribers.
private Subscriber[] _subscriberSnapshot = [];
// Replay ring buffer. Appended on the pump thread and queried from arbitrary
@@ -295,9 +300,10 @@ public sealed class SessionEventDistributor : IAsyncDisposable
/// (gRPC) subscribers and excludes the internal dashboard subscriber.
/// </summary>
/// <remarks>
/// Read from the copy-on-write snapshot rather than <c>ConcurrentDictionary.Count</c>
/// (which acquires every internal lock). The snapshot is rebuilt in the same
/// <c>_lifecycleLock</c> section that mutates the dictionary, so the two never diverge.
/// Read from the copy-on-write snapshot rather than the dictionary, because this
/// property is a lock-free reader and the dictionary may only be touched under
/// <c>_lifecycleLock</c>. The snapshot is rebuilt in the same <c>_lifecycleLock</c>
/// section that mutates the dictionary, so the two never diverge.
/// </remarks>
public int SubscriberCount => Volatile.Read(ref _subscriberSnapshot).Length;
@@ -648,7 +654,7 @@ public sealed class SessionEventDistributor : IAsyncDisposable
// register". A subscriber UNREGISTERED after the capture is still written to,
// and TryWrite on its completed channel returns false — from here that is
// indistinguishable from a real overflow. The window predates the
// copy-on-write array (ConcurrentDictionary.Values materialized its list up
// copy-on-write array (enumerating the dictionary materialized its values up
// front too) and its outcome is NOT benign, so telling a graceful unregister
// apart from a genuine overflow is OnSubscriberOverflow's job, not this
// loop's.
@@ -796,7 +802,7 @@ public sealed class SessionEventDistributor : IAsyncDisposable
{
lock (_lifecycleLock)
{
if (!_subscribers.TryRemove(subscriber.Id, out _))
if (!_subscribers.Remove(subscriber.Id))
{
return false;
}
@@ -808,7 +814,8 @@ public sealed class SessionEventDistributor : IAsyncDisposable
// Republishes the fan-out array from the current dictionary contents. MUST be called with
// _lifecycleLock held — holding that lock across the dictionary mutation and this rebuild is
// what keeps the array and the dictionary from diverging.
// what keeps the array and the dictionary from diverging, and it is also what makes the plain
// (non-concurrent) Dictionary safe: this enumeration never races a mutation.
private void RebuildSubscriberSnapshot()
=> Volatile.Write(ref _subscriberSnapshot, [.. _subscribers.Values]);