fix(dashboard): per-viewer share of the live-data advise set

The 256-tag LRU was global, so two Browse circuits on different areas evicted
each other's whole sets every poll. Tags now record the viewer that read them
last; each active viewer (read within 30s) owns MaxSubscribedTags/active
(floor 32). Eviction takes other viewers' excess over their share first
(inactive viewers' share is zero), then the reader's own untouched tags, and
never strips a neighbour within its share (soft cap). Single-viewer behaviour
is unchanged. IDashboardLiveDataService gains a viewerId ReadAsync overload.
This commit is contained in:
Joseph Doherty
2026-10-22 06:22:17 -04:00
parent 2a7d6facce
commit 5d4aefc99c
3 changed files with 279 additions and 31 deletions
@@ -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<DashboardLiveDataService> _logger;
private readonly TimeProvider _timeProvider;
private readonly SemaphoreSlim _gate = new(1, 1);
/// <summary>
@@ -70,6 +88,10 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
private readonly LinkedList<SubscribedTag> _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<string, ViewerState> _viewers = new(StringComparer.Ordinal);
private GatewaySession? _session;
private int _serverHandle;
private bool _disposed;
@@ -79,22 +101,31 @@ public sealed class DashboardLiveDataService : IDashboardLiveDataService, IAsync
/// <param name="alarmService">Gateway central alarm service.</param>
/// <param name="options">Gateway options supplying <c>Dashboard:ShowTagValues</c>.</param>
/// <param name="logger">Diagnostic logger.</param>
/// <param name="timeProvider">Clock for viewer activity; defaults to system time.</param>
public DashboardLiveDataService(
ISessionManager sessionManager,
IGatewayAlarmService alarmService,
IOptions<GatewayOptions> options,
ILogger<DashboardLiveDataService> logger)
ILogger<DashboardLiveDataService> 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;
}
/// <inheritdoc />
public Task<DashboardLiveReadResult> ReadAsync(
IReadOnlyCollection<string> tagAddresses,
CancellationToken cancellationToken) => ReadAsync(tagAddresses, viewerId: null, cancellationToken);
/// <inheritdoc />
public async Task<DashboardLiveReadResult> ReadAsync(
IReadOnlyCollection<string> 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<string> 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<SubscribeResult> subscribeResults = await session
.SubscribeBulkAsync(serverHandle, toSubscribe, cancellationToken)
.ConfigureAwait(false);
TrackSubscribed(toSubscribe, subscribeResults);
TrackSubscribed(toSubscribe, subscribeResults, viewer);
}
IReadOnlyList<BulkReadResult> 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<string> tagAddresses, out int justReadCount)
private string[] TouchAndCollectNewTags(
IReadOnlyCollection<string> tagAddresses,
string viewer,
out HashSet<string> justRead)
{
int touched = 0;
List<string> toSubscribe = [];
HashSet<string> 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<string> 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<int> evictedHandles = new(evictCount);
for (int i = 0; i < evictCount && _recency.Last is { } oldest; i++)
int share = PerViewerCap(now);
List<int> evictedHandles = [];
int evictCount = 0;
// Pass 1: other viewers over their share.
for (LinkedListNode<SubscribedTag>? node = _recency.Last; node is not null && overflow > 0;)
{
_recency.RemoveLast();
_subscribed.Remove(oldest.Value.TagAddress);
if (oldest.Value.ItemHandle != 0)
LinkedListNode<SubscribedTag>? 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<SubscribedTag>? node = _recency.Last; node is not null && overflow > 0;)
{
LinkedListNode<SubscribedTag>? 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<SubscribedTag> node, List<int> 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<string, ViewerState> 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<string> tagAddresses, IReadOnlyList<SubscribeResult> results)
private void TrackSubscribed(
IReadOnlyList<string> tagAddresses,
IReadOnlyList<SubscribeResult> results,
string viewer)
{
Dictionary<string, int> 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; }
}
}
@@ -20,6 +20,24 @@ public interface IDashboardLiveDataService
IReadOnlyCollection<string> tagAddresses,
CancellationToken cancellationToken);
/// <summary>
/// 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 <see cref="ReadAsync(IReadOnlyCollection{string}, CancellationToken)"/>.
/// </summary>
/// <param name="tagAddresses">Fully-qualified tag references to read.</param>
/// <param name="viewerId">
/// Stable id of the reading viewer for the life of its page; <see langword="null"/>
/// reads under one shared anonymous viewer.
/// </param>
/// <param name="cancellationToken">Token to cancel the read.</param>
/// <returns>The read result, or an error-bearing result on failure.</returns>
Task<DashboardLiveReadResult> ReadAsync(
IReadOnlyCollection<string> tagAddresses,
string? viewerId,
CancellationToken cancellationToken) => ReadAsync(tagAddresses, cancellationToken);
/// <summary>
/// Queries the currently-active alarm set for the dashboard session.
/// Never throws — failures are surfaced in
@@ -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);
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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);
}
/// <summary>
/// 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.
/// </summary>
/// <returns>A task that represents the asynchronous operation.</returns>
[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);
}
/// <summary>
/// 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<DashboardLiveDataService>.Instance);
NullLogger<DashboardLiveDataService>.Instance,
timeProvider);
}
private static string[] CreateTagAddresses(int count, string prefix = "Tank")