dohertj2/ScadaBridge
1
0
Fork 0
Code Issues 2 Pull Requests Actions Packages Projects Releases Wiki Activity

perf(dcl): coalesce MxGateway alarm stream reopens across a subscribe burst (review 2026-10-21 02/P3)

Browse Source
This commit is contained in:
Joseph Doherty
2026-10-22 06:44:13 -04:00
parent e6406c494c
commit d8d44e0b87
7 changed files with 189 additions and 36 deletions
Download Patch File Download Diff File Expand all files Collapse all files
src/ZB.MOM.WW.ScadaBridge.DataConnectionLayer/Adapters/MxGatewayDataConnection.cs
+90 -30
View File
@@ -50,6 +50,17 @@ public class MxGatewayDataConnection
private readonly Dictionary<string, string> _alarmSubSources = new(StringComparer.Ordinal);
private string _alarmStreamPrefix = string.Empty;
// Reopen coalescing (review 2026-10-21 02/P3). Every new source needs a fresh standing-state
// snapshot, but a burst of subscribes (a staggered deploy batch of alarm-bound instances)
// used to cost one gateway stream close/reopen — and one connection-wide snapshot replay to
// every alarm subscriber — PER source. Subscribes inside the window share ONE reopen on the
// union prefix. The ticket invalidates a scheduled reopen that a reconnect/teardown or a
// last-unsubscribe superseded. All guarded by _alarmLock.
private readonly TimeSpan _alarmStreamReopenDebounce;
private AlarmTransitionCallback? _alarmCallback;
private int _alarmReopenTicket;
private bool _alarmReopenPending;
// subscriptionId → (tagPath, callback) so the event loop can route updates by tag,
// plus tagPath → subscriptionId for reverse lookup. Concurrent because the event
// loop reads from a background thread while Subscribe/Unsubscribe mutate.
@@ -80,6 +91,11 @@ public class MxGatewayDataConnection
/// Maximum in-flight supervisory advise commands on the bulk-subscribe path when the
/// endpoint has no write-user context (<c>DataConnectionOptions.MxSupervisoryAdviseParallelism</c>).
/// </param>
/// <param name="alarmStreamReopenDebounce">
/// Window over which alarm subscribes are coalesced into one stream reopen
/// (<c>DataConnectionOptions.MxAlarmStreamReopenDebounce</c>). Zero reopens synchronously
/// per subscribe (the historical behaviour). Null uses the option's default.
/// </param>
/// <param name="clientNamePrefix">
/// Gateway client registration name used when the connection's own <c>ClientName</c> is blank
/// (<c>ScadaBridge:MxGateway:ClientNamePrefix</c>, default <c>scadabridge</c>). Null or blank
@@ -90,8 +106,11 @@ public class MxGatewayDataConnection
ILogger<MxGatewayDataConnection> logger,
ISecretResolver? secretResolver = null,
int supervisoryAdviseParallelism = 16,
string? clientNamePrefix = null)
string? clientNamePrefix = null,
TimeSpan? alarmStreamReopenDebounce = null)
{
var debounce = alarmStreamReopenDebounce ?? new DataConnectionOptions().MxAlarmStreamReopenDebounce;
_alarmStreamReopenDebounce = debounce < TimeSpan.Zero ? TimeSpan.Zero : debounce;
_clientFactory = clientFactory;
_logger = logger;
_secretResolver = secretResolver;
@@ -173,6 +192,7 @@ public class MxGatewayDataConnection
_alarmSubCount = 0;
_alarmSubSources.Clear();
_alarmStreamPrefix = string.Empty;
CancelPendingAlarmReopenLocked();
}
// New session: the previous session's items died with it. Drop their routing entries
@@ -311,6 +331,7 @@ public class MxGatewayDataConnection
_alarmSubCount = 0;
_alarmSubSources.Clear();
_alarmStreamPrefix = string.Empty;
CancelPendingAlarmReopenLocked();
}
if (_client is not null)
await _client.DisconnectAsync(cancellationToken);
@@ -399,51 +420,88 @@ public class MxGatewayDataConnection
{
_alarmSubCount++;
_alarmSubSources[subscriptionId] = sourceReference;
// The DataConnectionActor routes every transition by source reference, so the one
// shared feed needs a single callback; the most recent registration serves.
_alarmCallback = callback;
// Push the UNION of the subscribed source references to the gateway as the
// stream's single filter prefix. It is a bandwidth optimisation only — the
// MxGateway has no server-side CONDITION filter either, so conditionFilter is
// still not forwarded: the DataConnectionActor's per-source + per-condition-type
// gate stays authoritative, uniform with the OPC UA WhereClause stance.
// Even a covered new source needs a standing-state snapshot. Reopen
// the shared stream; DCL separately replays same-source co-subscribers.
// A new source outside the live prefix (or the first subscriber): (re)open the
// stream on the recomputed union. The source replays a fresh snapshot, which
// NativeAlarmActor already handles.
if (_alarmCts != null)
// Even a covered new source needs a standing-state snapshot, so every new source
// (re)opens the shared stream — coalesced across a burst (see the fields above);
// DCL separately replays same-source co-subscribers.
if (_alarmStreamReopenDebounce <= TimeSpan.Zero)
{
_alarmCts.Cancel();
_alarmCts.Dispose();
ReopenAlarmStreamLocked();
}
_alarmStreamPrefix = AlarmFilterPrefix.LongestCommonPrefix(_alarmSubSources.Values);
_alarmCts = new CancellationTokenSource();
var token = _alarmCts.Token;
var client = _client!;
var prefix = _alarmStreamPrefix.Length == 0 ? null : _alarmStreamPrefix;
var epoch = Guid.NewGuid();
void Forward(Commons.Types.Alarms.NativeAlarmTransition transition)
else if (!_alarmReopenPending)
{
lock (_alarmLock)
_alarmReopenPending = true;
var ticket = ++_alarmReopenTicket;
_ = Task.Delay(_alarmStreamReopenDebounce).ContinueWith(_ =>
{
if (token.IsCancellationRequested) return;
if (transition.Kind == AlarmTransitionKind.SnapshotStart) epoch = Guid.NewGuid();
callback(transition.Kind is AlarmTransitionKind.SnapshotStart or AlarmTransitionKind.Snapshot or AlarmTransitionKind.SnapshotComplete
? transition with { SnapshotId = epoch } : transition);
}
lock (_alarmLock)
{
if (ticket != _alarmReopenTicket || !_alarmReopenPending)
return; // superseded by a reconnect, teardown or last unsubscribe
_alarmReopenPending = false;
if (_alarmSubCount > 0 && _client is not null)
ReopenAlarmStreamLocked();
}
}, TaskScheduler.Default);
}
_ = Task.Run(async () =>
{
Forward(MxGatewayAlarmMapper.SnapshotComplete() with { Kind = AlarmTransitionKind.SnapshotStart });
if (token.IsCancellationRequested) return;
await client.RunAlarmStreamAsync(prefix, Forward, token);
}, token);
}
return Task.FromResult(subscriptionId);
}
/// <summary>
/// (Re)opens the shared alarm stream on the union prefix of the subscribed sources. The
/// source replays a fresh snapshot, which NativeAlarmActor already handles. Callers hold
/// <see cref="_alarmLock"/>.
/// </summary>
private void ReopenAlarmStreamLocked()
{
if (_alarmCts != null)
{
_alarmCts.Cancel();
_alarmCts.Dispose();
}
_alarmStreamPrefix = AlarmFilterPrefix.LongestCommonPrefix(_alarmSubSources.Values);
_alarmCts = new CancellationTokenSource();
var token = _alarmCts.Token;
var client = _client!;
var callback = _alarmCallback!;
var prefix = _alarmStreamPrefix.Length == 0 ? null : _alarmStreamPrefix;
var epoch = Guid.NewGuid();
void Forward(Commons.Types.Alarms.NativeAlarmTransition transition)
{
lock (_alarmLock)
{
if (token.IsCancellationRequested) return;
if (transition.Kind == AlarmTransitionKind.SnapshotStart) epoch = Guid.NewGuid();
callback(transition.Kind is AlarmTransitionKind.SnapshotStart or AlarmTransitionKind.Snapshot or AlarmTransitionKind.SnapshotComplete
? transition with { SnapshotId = epoch } : transition);
}
}
_ = Task.Run(async () =>
{
Forward(MxGatewayAlarmMapper.SnapshotComplete() with { Kind = AlarmTransitionKind.SnapshotStart });
if (token.IsCancellationRequested) return;
await client.RunAlarmStreamAsync(prefix, Forward, token);
}, token);
}
/// <summary>Invalidates a scheduled coalesced reopen. Callers hold <see cref="_alarmLock"/>.</summary>
private void CancelPendingAlarmReopenLocked()
{
_alarmReopenPending = false;
_alarmReopenTicket++;
}
/// <inheritdoc />
public Task UnsubscribeAlarmsAsync(string subscriptionId, CancellationToken cancellationToken = default)
{
@@ -464,6 +522,7 @@ public class MxGatewayDataConnection
_alarmCts?.Dispose();
_alarmCts = null;
_alarmStreamPrefix = string.Empty;
CancelPendingAlarmReopenLocked();
}
}
return Task.CompletedTask;
@@ -541,6 +600,7 @@ public class MxGatewayDataConnection
_alarmSubCount = 0;
_alarmSubSources.Clear();
_alarmStreamPrefix = string.Empty;
CancelPendingAlarmReopenLocked();
}
if (_client is not null)
await _client.DisposeAsync();