diff --git a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs index a7b653a..e2bc100 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Sessions/SessionEventDistributor.cs @@ -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 /// EventStreamService.ProduceEventsAsync ordering. /// /// -/// Concurrency. The subscriber set is a -/// keyed by a monotonic id, used -/// for keyed add/remove only. Fan-out does NOT enumerate the dictionary: every -/// mutation (, , lease -/// disposal, overflow disconnect) happens inside the _lifecycleLock critical -/// section and rebuilds an immutable copy-on-write Subscriber[] snapshot, -/// which the pump reads once per event. This matters because -/// ConcurrentDictionary.Values is a PROPERTY that acquires every internal -/// lock and materializes a fresh List 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 +/// Concurrency. The subscriber set is a plain +/// keyed by a monotonic id, used for keyed +/// add/remove only. It needs no concurrent collection type because it is never +/// touched outside the _lifecycleLock critical section: every mutation +/// (, , 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 +/// Subscriber[] snapshot inside the same section. The lock-free readers see only that snapshot, +/// never the dictionary: the pump reads it once per event and +/// 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 /// await. 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 _logger; private readonly TimeProvider _timeProvider; - private readonly ConcurrentDictionary _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 _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. /// /// - /// Read from the copy-on-write snapshot rather than ConcurrentDictionary.Count - /// (which acquires every internal lock). The snapshot is rebuilt in the same - /// _lifecycleLock 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 + /// _lifecycleLock. The snapshot is rebuilt in the same _lifecycleLock + /// section that mutates the dictionary, so the two never diverge. /// 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]);