diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardLiveDataService.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardLiveDataService.cs index 2239c65..a1aa716 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardLiveDataService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/DashboardLiveDataService.cs @@ -34,20 +34,38 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync // // The bound is per-read, not absolute: a read may never evict a tag it is itself // about to return, so a single read of more distinct tags than the cap leaves the - // set that large. The invariant EvictForAsync actually maintains is + // set that large. With a single viewer the invariant EvictForAsync maintains is // // |advise set| after a read <= max(MaxSubscribedTags, distinct tags in that read) // - // and any overshoot is squeezed back out by the next read that subscribes a tag + // (with several active viewers the per-viewer share below can hold it higher, by + // at most what the other viewers hold within their shares) and any overshoot is squeezed back out by the next read that subscribes a tag // (see EvictForAsync). A browse page requests far fewer tags than the cap, so in // practice the set settles at MaxSubscribedTags. private const int MaxSubscribedTags = 256; + // Per-viewer soft cap: MaxSubscribedTags is shared by every active viewer (Browse + // circuit), so each owns at most MaxSubscribedTags / activeViewers tags before its + // own least-recently-read tags become the first eviction candidates — never fewer + // than this floor. Without the split, two circuits browsing different areas + // thrashed the single global LRU: each read evicted the other circuit's whole set, + // which then re-advised it two seconds later. + private const int MinTagsPerViewer = 32; + + // A viewer that has not read within this window no longer counts as active: its + // share is released and its tags become the first eviction candidates. Generous + // against the Browse page's 2 s poll, so a briefly paused circuit keeps its share. + private static readonly TimeSpan ActiveViewerWindow = TimeSpan.FromSeconds(30); + + // Viewer id for callers that do not identify themselves. + private const string SharedViewerId = ""; + private static readonly TimeSpan ReadTimeout = TimeSpan.FromSeconds(5); private readonly ISessionManager _sessionManager; private readonly IGatewayAlarmService _alarmService; private readonly ILogger _logger; + private readonly TimeProvider _timeProvider; private readonly SemaphoreSlim _gate = new(1, 1); /// @@ -70,6 +88,10 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync private readonly LinkedList _recency = new(); + // Per-viewer ownership and activity. A tag is owned by the viewer that read it + // last; only touched under _gate. + private readonly Dictionary _viewers = new(StringComparer.Ordinal); + private GatewaySession? _session; private int _serverHandle; private bool _disposed; @@ -79,22 +101,31 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync /// Gateway central alarm service. /// Gateway options supplying Dashboard:ShowTagValues. /// Diagnostic logger. + /// Clock for viewer activity; defaults to system time. public DashboardLiveDataService( ISessionManager sessionManager, IGatewayAlarmService alarmService, IOptions options, - ILogger logger) + ILogger logger, + TimeProvider? timeProvider = null) { ArgumentNullException.ThrowIfNull(options); _sessionManager = sessionManager ?? throw new ArgumentNullException(nameof(sessionManager)); _alarmService = alarmService ?? throw new ArgumentNullException(nameof(alarmService)); _logger = logger ?? throw new ArgumentNullException(nameof(logger)); _showTagValues = options.Value.Dashboard.ShowTagValues; + _timeProvider = timeProvider ?? TimeProvider.System; } + /// + public Task ReadAsync( + IReadOnlyCollection tagAddresses, + CancellationToken cancellationToken) => ReadAsync(tagAddresses, viewerId: null, cancellationToken); + /// public async Task ReadAsync( IReadOnlyCollection tagAddresses, + string? viewerId, CancellationToken cancellationToken) { ArgumentNullException.ThrowIfNull(tagAddresses); @@ -109,15 +140,19 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync (GatewaySession session, int serverHandle) = await EnsureReadyAsync(cancellationToken) .ConfigureAwait(false); - string[] toSubscribe = TouchAndCollectNewTags(tagAddresses, out int justReadCount); + string viewer = viewerId ?? SharedViewerId; + DateTimeOffset now = _timeProvider.GetUtcNow(); + TouchViewer(viewer, now); + + string[] toSubscribe = TouchAndCollectNewTags(tagAddresses, viewer, out HashSet justRead); if (toSubscribe.Length > 0) { - await EvictForAsync(session, serverHandle, toSubscribe.Length, justReadCount, cancellationToken) + await EvictForAsync(session, serverHandle, toSubscribe.Length, justRead, viewer, now, cancellationToken) .ConfigureAwait(false); IReadOnlyList subscribeResults = await session .SubscribeBulkAsync(serverHandle, toSubscribe, cancellationToken) .ConfigureAwait(false); - TrackSubscribed(toSubscribe, subscribeResults); + TrackSubscribed(toSubscribe, subscribeResults, viewer); } IReadOnlyList results = await session @@ -175,18 +210,20 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync } // Promotes every already-advised tag in this read to the front of the recency - // list and returns the tags that still need subscribing (distinct, in request - // order). `justReadCount` is how many distinct tags of this read were already - // advised — they now occupy the front of the list and must never be evicted to - // make room for the same read's new tags. Callers must hold _gate. + // list, hands its ownership to the reading viewer, and returns the tags that + // still need subscribing (distinct, in request order). `justRead` is every + // distinct tag of this read — none of them may be evicted to make room for the + // same read's new tags. Callers must hold _gate. // // Every tag of one read is equally recently read; the recency list needs a total // order anyway, so the whole service uses one tie-break: later in the request wins. // Promoting in request order gives that here, and TrackSubscribed inserts new tags // the same way. - private string[] TouchAndCollectNewTags(IReadOnlyCollection tagAddresses, out int justReadCount) + private string[] TouchAndCollectNewTags( + IReadOnlyCollection tagAddresses, + string viewer, + out HashSet justRead) { - int touched = 0; List toSubscribe = []; HashSet seen = new(StringComparer.OrdinalIgnoreCase); @@ -200,10 +237,14 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync _recency.AddFirst(node); } - if (seen.Add(tag)) + if (!string.Equals(node.Value.Owner, viewer, StringComparison.Ordinal)) { - touched++; + AdjustOwned(node.Value.Owner, -1); + AdjustOwned(viewer, +1); + node.Value = node.Value with { Owner = viewer }; } + + seen.Add(tag); } else if (seen.Add(tag)) { @@ -211,7 +252,7 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync } } - justReadCount = touched; + justRead = seen; return [.. toSubscribe]; } @@ -221,7 +262,17 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync // from tracking regardless, and the session-invalidation path already handles // gateway/worker drift. Callers must hold _gate. // - // Eviction stops at the tags this read just touched (`justReadCount`), so a read + // Victims are chosen in two passes, each least-recently-read first: + // 1. tags of OTHER viewers that are over their share (PerViewerCap; an + // inactive viewer's share is zero); + // 2. the reading viewer's own tags that this read did not touch. + // A viewer within its share is never evicted by someone else's read, so the + // cap is soft: when every other viewer sits within its share and the reader's + // own older tags run out, the set ends over MaxSubscribedTags rather than + // stripping a neighbour. With one viewer, pass 1 is empty and pass 2 is the + // plain global LRU. + // + // Eviction never touches the tags this read just touched (`justRead`), so a read // whose own distinct tags outnumber the cap ends over it — see MaxSubscribedTags // for the exact invariant. That overshoot is not sticky: the next read that // subscribes anything computes `overflow` against the oversized set and evicts the @@ -236,26 +287,56 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync GatewaySession session, int serverHandle, int incomingCount, - int justReadCount, + HashSet justRead, + string reader, + DateTimeOffset now, CancellationToken cancellationToken) { int overflow = _subscribed.Count + incomingCount - MaxSubscribedTags; - int evictable = _subscribed.Count - justReadCount; - int evictCount = Math.Min(overflow, evictable); - if (evictCount <= 0) + if (overflow <= 0) { return; } - List evictedHandles = new(evictCount); - for (int i = 0; i < evictCount && _recency.Last is { } oldest; i++) + int share = PerViewerCap(now); + List evictedHandles = []; + int evictCount = 0; + + // Pass 1: other viewers over their share. + for (LinkedListNode? node = _recency.Last; node is not null && overflow > 0;) { - _recency.RemoveLast(); - _subscribed.Remove(oldest.Value.TagAddress); - if (oldest.Value.ItemHandle != 0) + LinkedListNode? previous = node.Previous; + string owner = node.Value.Owner; + if (!string.Equals(owner, reader, StringComparison.Ordinal) + && !justRead.Contains(node.Value.TagAddress) + && OwnedCount(owner) > (IsActive(owner, now) ? share : 0)) { - evictedHandles.Add(oldest.Value.ItemHandle); + Evict(node, evictedHandles); + overflow--; + evictCount++; } + + node = previous; + } + + // Pass 2: the reader's own tags this read did not touch. + for (LinkedListNode? node = _recency.Last; node is not null && overflow > 0;) + { + LinkedListNode? previous = node.Previous; + if (string.Equals(node.Value.Owner, reader, StringComparison.Ordinal) + && !justRead.Contains(node.Value.TagAddress)) + { + Evict(node, evictedHandles); + overflow--; + evictCount++; + } + + node = previous; + } + + if (evictCount == 0) + { + return; } _logger.LogDebug( @@ -282,12 +363,81 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync } } + // Removes one tag from tracking, releasing its owner's count and collecting its + // item handle for the batch unadvise. Callers must hold _gate. + private void Evict(LinkedListNode node, List evictedHandles) + { + _recency.Remove(node); + _subscribed.Remove(node.Value.TagAddress); + AdjustOwned(node.Value.Owner, -1); + if (node.Value.ItemHandle != 0) + { + evictedHandles.Add(node.Value.ItemHandle); + } + } + + // Each active viewer's share of MaxSubscribedTags, floored at MinTagsPerViewer. + // Callers must hold _gate. + private int PerViewerCap(DateTimeOffset now) + { + int active = _viewers.Values.Count(state => now - state.LastReadAt <= ActiveViewerWindow); + return Math.Max(MinTagsPerViewer, MaxSubscribedTags / Math.Max(1, active)); + } + + private bool IsActive(string viewer, DateTimeOffset now) => + _viewers.TryGetValue(viewer, out ViewerState? state) && now - state.LastReadAt <= ActiveViewerWindow; + + private int OwnedCount(string viewer) => + _viewers.TryGetValue(viewer, out ViewerState? state) ? state.OwnedCount : 0; + + // Records a read by `viewer` and forgets viewers that are both inactive and own + // nothing, so the map does not grow one entry per Browse page ever opened. + // Callers must hold _gate. + private void TouchViewer(string viewer, DateTimeOffset now) + { + if (!_viewers.TryGetValue(viewer, out ViewerState? state)) + { + state = new ViewerState(); + _viewers[viewer] = state; + } + + state.LastReadAt = now; + + foreach (KeyValuePair entry in _viewers.ToArray()) + { + if (entry.Value.OwnedCount <= 0 && now - entry.Value.LastReadAt > ActiveViewerWindow) + { + _viewers.Remove(entry.Key); + } + } + } + + // Callers must hold _gate. + private void AdjustOwned(string viewer, int delta) + { + if (!_viewers.TryGetValue(viewer, out ViewerState? state)) + { + if (delta <= 0) + { + return; + } + + state = new ViewerState { LastReadAt = DateTimeOffset.MinValue }; + _viewers[viewer] = state; + } + + state.OwnedCount = Math.Max(0, state.OwnedCount + delta); + } + // Records the freshly advised tags as the most recently read, keeping each // tag's item handle so eviction can unadvise it. Tags the worker failed to // advise are still tracked (matching the pre-cap behaviour of not retrying // them on every read) but carry no handle, so eviction just forgets them. // Callers must hold _gate. - private void TrackSubscribed(IReadOnlyList tagAddresses, IReadOnlyList results) + private void TrackSubscribed( + IReadOnlyList tagAddresses, + IReadOnlyList results, + string viewer) { Dictionary handles = new(results.Count, StringComparer.OrdinalIgnoreCase); foreach (SubscribeResult result in results) @@ -303,7 +453,8 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync foreach (string tag in tagAddresses) { handles.TryGetValue(tag, out int itemHandle); - _subscribed[tag] = _recency.AddFirst(new SubscribedTag(tag, itemHandle)); + _subscribed[tag] = _recency.AddFirst(new SubscribedTag(tag, itemHandle, viewer)); + AdjustOwned(viewer, +1); } } @@ -314,6 +465,10 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync { _subscribed.Clear(); _recency.Clear(); + foreach (ViewerState state in _viewers.Values) + { + state.OwnedCount = 0; + } } // Returns a Ready session + its Register server handle, opening a fresh @@ -424,5 +579,14 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync // One entry of the advise set. ItemHandle is the handle the worker bound for // the tag, or 0 when the subscribe failed and there is nothing to unadvise. - private readonly record struct SubscribedTag(string TagAddress, int ItemHandle); + // Owner is the viewer that read the tag most recently. + private readonly record struct SubscribedTag(string TagAddress, int ItemHandle, string Owner); + + // One viewer's slice of the advise set. + private sealed class ViewerState + { + public int OwnedCount { get; set; } + + public DateTimeOffset LastReadAt { get; set; } + } } diff --git a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/IDashboardLiveDataService.cs b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/IDashboardLiveDataService.cs index 698cc5b..e37306a 100644 --- a/src/ZB.MOM.WW.MxGateway.Server/Dashboard/IDashboardLiveDataService.cs +++ b/src/ZB.MOM.WW.MxGateway.Server/Dashboard/IDashboardLiveDataService.cs @@ -20,6 +20,24 @@ public interface IDashboardLiveDataService IReadOnlyCollection tagAddresses, CancellationToken cancellationToken); + /// + /// Reads the supplied tags on behalf of one viewer (a Browse page circuit). + /// The shared advise set is budgeted per active viewer, so the viewer id is + /// what stops one circuit's reads from evicting another circuit's whole set. + /// Never throws, as . + /// + /// Fully-qualified tag references to read. + /// + /// Stable id of the reading viewer for the life of its page; + /// reads under one shared anonymous viewer. + /// + /// Token to cancel the read. + /// The read result, or an error-bearing result on failure. + Task ReadAsync( + IReadOnlyCollection tagAddresses, + string? viewerId, + CancellationToken cancellationToken) => ReadAsync(tagAddresses, cancellationToken); + /// /// Queries the currently-active alarm set for the dashboard session. /// Never throws — failures are surfaced in diff --git a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardLiveDataServiceTests.cs b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardLiveDataServiceTests.cs index 820e29b..af52910 100644 --- a/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardLiveDataServiceTests.cs +++ b/src/ZB.MOM.WW.MxGateway.Tests/Gateway/Dashboard/DashboardLiveDataServiceTests.cs @@ -2,6 +2,7 @@ using System.Diagnostics.CodeAnalysis; using System.Runtime.CompilerServices; using Microsoft.Extensions.Logging.Abstractions; using Microsoft.Extensions.Options; +using Microsoft.Extensions.Time.Testing; using ZB.MOM.WW.MxGateway.Contracts.Proto; using ZB.MOM.WW.MxGateway.Server.Configuration; using ZB.MOM.WW.MxGateway.Server.Dashboard; @@ -135,6 +136,69 @@ public sealed class DashboardLiveDataServiceTests Assert.Equal(5, worker.UnsubscribedHandles.Count); } + /// + /// Verifies the per-viewer share: with two active circuits each owns half the cap, + /// so a second circuit's large read evicts only the first circuit's excess above + /// its share — never its whole set — and keeps every tag it read itself. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReadAsync_WithTwoViewers_EvictsOnlyTheOtherViewersExcessOverItsShare() + { + RecordingWorkerClient worker = new(); + await using FakeSessionManager sessionManager = new(worker); + await using DashboardLiveDataService service = CreateService(sessionManager); + + string[] first = CreateTagAddresses(200, "First"); + string[] second = CreateTagAddresses(200, "Second"); + await service.ReadAsync(first, "circuit-a", CancellationToken.None); + DashboardLiveReadResult result = await service.ReadAsync(second, "circuit-b", CancellationToken.None); + + Assert.Null(result.Error); + + // Two active viewers -> share of 128 each. circuit-a drops its 72 least-recently-read + // tags and keeps 128; under the old global LRU it lost 144 (400 - 256). + int share = MaxSubscribedTags / 2; + Assert.Equal(first[..(first.Length - share)].Select(worker.HandleFor), worker.UnsubscribedHandles); + Assert.DoesNotContain(worker.UnsubscribedHandles, handle => second.Select(worker.HandleFor).Contains(handle)); + + // circuit-a re-reading its retained tags subscribes nothing new and evicts nothing. + worker.UnsubscribedHandles.Clear(); + int subscribedBefore = worker.SubscribedTags.Count; + await service.ReadAsync(first[^share..], "circuit-a", CancellationToken.None); + Assert.Equal(subscribedBefore, worker.SubscribedTags.Count); + Assert.Empty(worker.UnsubscribedHandles); + } + + /// + /// Verifies an inactive viewer (a closed circuit that stopped polling) loses its + /// share: its tags are the first eviction candidates once the activity window passes. + /// + /// A task that represents the asynchronous operation. + [Fact] + public async Task ReadAsync_WhenAViewerGoesInactive_ItsTagsAreEvictedFirst() + { + RecordingWorkerClient worker = new(); + await using FakeSessionManager sessionManager = new(worker); + FakeTimeProvider clock = new(); + await using DashboardLiveDataService service = CreateService(sessionManager, timeProvider: clock); + + string[] stale = CreateTagAddresses(100, "Stale"); + string[] live = CreateTagAddresses(156, "Live"); + await service.ReadAsync(live, "open-circuit", CancellationToken.None); + await service.ReadAsync(stale, "closed-circuit", CancellationToken.None); + Assert.Empty(worker.UnsubscribedHandles); + + clock.Advance(TimeSpan.FromMinutes(1)); + + // The stale tags were read after the live ones, so a global LRU would evict live[10] + // (the oldest tag this read does not touch). The stale owner is inactive, though, so + // its share is zero and its oldest tag goes first. + await service.ReadAsync([.. live[..10], "Live.New.PV"], "open-circuit", CancellationToken.None); + + Assert.Equal([worker.HandleFor(stale[0])], worker.UnsubscribedHandles); + } + /// /// Verifies a tag the worker failed to advise still occupies a slot but is evicted /// without any unsubscribe command — there is no item handle to unadvise. @@ -280,7 +344,8 @@ public sealed class DashboardLiveDataServiceTests private static DashboardLiveDataService CreateService( ISessionManager sessionManager, - bool showTagValues = false) + bool showTagValues = false, + TimeProvider? timeProvider = null) { GatewayOptions gatewayOptions = new() { @@ -291,7 +356,8 @@ public sealed class DashboardLiveDataServiceTests sessionManager, new FakeGatewayAlarmService(), Options.Create(gatewayOptions), - NullLogger.Instance); + NullLogger.Instance, + timeProvider); } private static string[] CreateTagAddresses(int count, string prefix = "Tank")