refactor(virtualtags): retire dormant VirtualTagEngine/Source/Definition + TimerTriggerScheduler (#603)
TimerTriggerScheduler was welded to the engine and has no production consumer. Claude-Session: https://claude.ai/code/session_01Ef2ozxS7QPanGT8uoATbTe
This commit is contained in:
Joseph Doherty
@@ -5,8 +5,7 @@ namespace ZB.MOM.WW.OtOpcUa.Commons.Engines;
|
||||
/// <summary>
|
||||
/// Abstraction over the compiled virtual-tag expression engine. Runtime consumes this so
|
||||
/// <see cref="VirtualTagActor"/> can stay free of Roslyn / scripting machinery and the
|
||||
/// production wiring binds an adapter over <c>VirtualTagEngine</c> from
|
||||
/// <c>Core.VirtualTags</c>.
|
||||
/// production wiring binds <c>RoslynVirtualTagEvaluator</c> (Host).
|
||||
/// </summary>
|
||||
public interface IVirtualTagEvaluator
|
||||
{
|
||||
@@ -98,7 +97,7 @@ public static class ScriptEvalFailureReasons
|
||||
}
|
||||
|
||||
/// <summary>Returns <see cref="VirtualTagEvalResult.NoChange"/> from every call. Bound by default
|
||||
/// when the production <c>VirtualTagEngine</c> adapter hasn't been registered (Mac dev, tests).</summary>
|
||||
/// when the production <c>RoslynVirtualTagEvaluator</c> hasn't been registered (Mac dev, tests).</summary>
|
||||
public sealed class NullVirtualTagEvaluator : IVirtualTagEvaluator
|
||||
{
|
||||
public static readonly NullVirtualTagEvaluator Instance = new();
|
||||
|
||||
@@ -39,7 +39,7 @@ namespace ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
/// </param>
|
||||
/// <param name="VirtualTagId">
|
||||
/// Set when <paramref name="Source"/> is <see cref="NodeSourceKind.Virtual"/> — stable
|
||||
/// logical id the VirtualTagEngine addresses by. Null otherwise.
|
||||
/// logical id the virtual-tag runtime addresses by. Null otherwise.
|
||||
/// </param>
|
||||
/// <param name="ScriptedAlarmId">
|
||||
/// Set when <paramref name="Source"/> is <see cref="NodeSourceKind.ScriptedAlarm"/> —
|
||||
@@ -78,7 +78,7 @@ public sealed record DriverAttributeInfo(
|
||||
/// Discriminates which runtime subsystem owns this node's Read/Write/
|
||||
/// Subscribe dispatch. <c>Driver</c> = a real IDriver capability surface;
|
||||
/// <c>Virtual</c> = a Phase 7 <see cref="DriverAttributeInfo"/>.VirtualTagId'd tag
|
||||
/// computed by the VirtualTagEngine; <c>ScriptedAlarm</c> = a scripted Part 9 alarm
|
||||
/// computed by the virtual-tag runtime; <c>ScriptedAlarm</c> = a scripted Part 9 alarm
|
||||
/// materialized by the ScriptedAlarmEngine.
|
||||
/// </summary>
|
||||
public enum NodeSourceKind
|
||||
|
||||
@@ -8,9 +8,7 @@ namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
||||
/// Per-evaluation <see cref="ScriptContext"/> for a virtual-tag script. Reads come
|
||||
/// out of the engine's last-known-value cache (driver tags updated via the
|
||||
/// <c>ITagUpstreamSource</c> subscription, virtual tags updated by prior
|
||||
/// evaluations). Writes route through <c>VirtualTagEngine</c>'s
|
||||
/// <c>OnScriptSetVirtualTag</c> callback so cross-tag write side effects still
|
||||
/// participate in change-trigger cascades (via the engine's <c>CascadeAsync</c>).
|
||||
/// evaluations). Writes route through the host-supplied set-virtual-tag callback.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
|
||||
@@ -15,11 +15,8 @@ namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
||||
/// from a last-known-value cache populated by the subscription callbacks.
|
||||
/// </para>
|
||||
/// <para>
|
||||
/// The subscription path feeds <see cref="VirtualTagEngine"/>'s
|
||||
/// <c>OnUpstreamChange</c> callback, which updates the engine's value cache and
|
||||
/// schedules <c>CascadeAsync</c> to re-evaluate every change-driven dependent in
|
||||
/// topological order. One subscription per distinct upstream tag path; the
|
||||
/// engine tracks the mapping itself.
|
||||
/// The subscription path feeds the consumer's change callback (scripted alarms via
|
||||
/// <c>DependencyMuxTagUpstreamSource</c>). One subscription per distinct upstream tag path.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public interface ITagUpstreamSource
|
||||
|
||||
@@ -1,162 +0,0 @@
|
||||
using Serilog;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
||||
|
||||
/// <summary>
|
||||
/// Periodic re-evaluation scheduler for tags with a non-null
|
||||
/// <see cref="VirtualTagDefinition.TimerInterval"/>. Independent of the
|
||||
/// change-trigger path — a tag can be timer-only, change-only, or both. One
|
||||
/// <see cref="System.Threading.Timer"/> per interval-group keeps the wire count
|
||||
/// low regardless of tag count.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// Each timer group carries a per-group in-flight flag (see
|
||||
/// <c>TickGroup.InFlight</c>). When the timer fires while a tick for the same
|
||||
/// group is still running, the new callback skips the work and increments
|
||||
/// <see cref="SkippedTickCount"/> rather than blocking a thread-pool thread on
|
||||
/// the engine's evaluation gate. This bounds the work outstanding at one tick
|
||||
/// per group, regardless of how long an individual evaluation takes.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class TimerTriggerScheduler : IDisposable
|
||||
{
|
||||
private readonly VirtualTagEngine _engine;
|
||||
private readonly ILogger _logger;
|
||||
private readonly List<Timer> _timers = [];
|
||||
private readonly List<TickGroup> _groups = [];
|
||||
private readonly CancellationTokenSource _cts = new();
|
||||
private long _skippedTickCount;
|
||||
private bool _disposed;
|
||||
private bool _started;
|
||||
|
||||
/// <summary>Initializes a new instance of the <see cref="TimerTriggerScheduler"/> class.</summary>
|
||||
/// <param name="engine">The virtual tag engine to trigger evaluations on.</param>
|
||||
/// <param name="logger">A logger for diagnostic output.</param>
|
||||
public TimerTriggerScheduler(VirtualTagEngine engine, ILogger logger)
|
||||
{
|
||||
_engine = engine ?? throw new ArgumentNullException(nameof(engine));
|
||||
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Diagnostic counter: number of timer callbacks that skipped their work because
|
||||
/// the prior tick for the same group was still running. Exposed for tests +
|
||||
/// operational metrics. Monotonic; never resets.
|
||||
/// </summary>
|
||||
public long SkippedTickCount => Interlocked.Read(ref _skippedTickCount);
|
||||
|
||||
/// <summary>
|
||||
/// Stand up one <see cref="Timer"/> per unique interval. All tags with
|
||||
/// matching interval share a timer; each tick triggers re-evaluation of the
|
||||
/// group in topological order so cascades are consistent with change-triggered
|
||||
/// behavior.
|
||||
/// </summary>
|
||||
/// <param name="definitions">The virtual tag definitions to schedule timers for.</param>
|
||||
/// <exception cref="InvalidOperationException">Thrown when <c>Start</c> is called more than once on the same instance.</exception>
|
||||
public void Start(IReadOnlyList<VirtualTagDefinition> definitions)
|
||||
{
|
||||
if (_disposed) throw new ObjectDisposedException(nameof(TimerTriggerScheduler));
|
||||
if (_started) throw new InvalidOperationException(
|
||||
"TimerTriggerScheduler.Start has already been called. Create a new instance for each Load cycle.");
|
||||
_started = true;
|
||||
|
||||
var byInterval = definitions
|
||||
.Where(d => d.TimerInterval.HasValue && d.TimerInterval.Value > TimeSpan.Zero)
|
||||
.GroupBy(d => d.TimerInterval!.Value);
|
||||
|
||||
foreach (var group in byInterval)
|
||||
{
|
||||
var paths = group.Select(d => d.Path).ToArray();
|
||||
var interval = group.Key;
|
||||
var ctx = new TickGroup(paths);
|
||||
_groups.Add(ctx);
|
||||
var timer = new Timer(_ => OnTimer(ctx), null, interval, interval);
|
||||
_timers.Add(timer);
|
||||
_logger.Information("TimerTriggerScheduler: {TagCount} tag(s) on {Interval} cadence",
|
||||
paths.Length, interval);
|
||||
}
|
||||
}
|
||||
|
||||
private void OnTimer(TickGroup ctx)
|
||||
{
|
||||
if (_cts.IsCancellationRequested) return;
|
||||
|
||||
// Skip the tick when the prior one for this group is still running. Without
|
||||
// this guard a slow evaluation (or one waiting on the engine's _evalGate) would
|
||||
// cause subsequent timer callbacks to each pin a thread-pool thread on the
|
||||
// gate, compounding under high tick rates.
|
||||
if (Interlocked.CompareExchange(ref ctx.InFlight, 1, 0) != 0)
|
||||
{
|
||||
Interlocked.Increment(ref _skippedTickCount);
|
||||
return;
|
||||
}
|
||||
|
||||
// Run async without blocking the timer's pool-thread callback. The task is
|
||||
// fire-and-forget — failures are logged inside RunTickAsync; the InFlight flag
|
||||
// is reset in the finally block so the next tick can proceed.
|
||||
_ = RunTickAsync(ctx);
|
||||
}
|
||||
|
||||
private async Task RunTickAsync(TickGroup ctx)
|
||||
{
|
||||
try
|
||||
{
|
||||
foreach (var p in ctx.Paths)
|
||||
{
|
||||
if (_cts.IsCancellationRequested) return;
|
||||
try
|
||||
{
|
||||
await _engine.EvaluateOneAsync(p, _cts.Token).ConfigureAwait(false);
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
return;
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_logger.Error(ex, "TimerTriggerScheduler evaluate failed for {Path}", p);
|
||||
}
|
||||
}
|
||||
}
|
||||
finally
|
||||
{
|
||||
Interlocked.Exchange(ref ctx.InFlight, 0);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Releases all timers and disposes the scheduler's resources.</summary>
|
||||
public void Dispose()
|
||||
{
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
_cts.Cancel();
|
||||
foreach (var t in _timers)
|
||||
{
|
||||
try { t.Dispose(); } catch { }
|
||||
}
|
||||
_timers.Clear();
|
||||
_groups.Clear();
|
||||
_cts.Dispose();
|
||||
}
|
||||
|
||||
/// <summary>Groups tags that share the same timer interval.</summary>
|
||||
private sealed class TickGroup
|
||||
{
|
||||
// 0 = idle, 1 = a tick is currently running (or queued) for this group. Use
|
||||
// Interlocked.CompareExchange so a timer callback observes a consistent "is the
|
||||
// prior tick still running" answer without taking a lock.
|
||||
/// <summary>Flag indicating whether a tick is currently in flight for this group.</summary>
|
||||
public int InFlight;
|
||||
|
||||
/// <summary>Gets the list of tag paths in this group.</summary>
|
||||
public IReadOnlyList<string> Paths { get; }
|
||||
|
||||
/// <summary>Initializes a new instance of the <see cref="TickGroup"/> class.</summary>
|
||||
/// <param name="paths">The tag paths in this timer group.</param>
|
||||
public TickGroup(IReadOnlyList<string> paths)
|
||||
{
|
||||
Paths = paths;
|
||||
}
|
||||
}
|
||||
}
|
||||
@@ -1,41 +0,0 @@
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
||||
|
||||
/// <summary>
|
||||
/// Operator-authored virtual-tag configuration row. Phase 7 Stream E (config DB
|
||||
/// schema) materializes these from the <c>VirtualTag</c> + <c>Script</c> tables on
|
||||
/// publish; the engine ingests a list of them at load time.
|
||||
/// </summary>
|
||||
/// <param name="Path">
|
||||
/// UNS tag path — <c>Enterprise/Site/Area/Line/Equipment/TagName</c>. Used both as
|
||||
/// the engine's internal id and the OPC UA browse path.
|
||||
/// </param>
|
||||
/// <param name="DataType">
|
||||
/// Expected return type. The evaluator coerces the script's return value to this
|
||||
/// type before publishing; mismatch surfaces as <c>BadTypeMismatch</c> quality on
|
||||
/// the tag.
|
||||
/// </param>
|
||||
/// <param name="ScriptSource">Roslyn C# script source. Must compile under <c>ScriptSandbox</c>.</param>
|
||||
/// <param name="ChangeTriggered">
|
||||
/// True if any input tag's change (value / status / timestamp delta) should trigger
|
||||
/// re-evaluation. Operator picks per tag — usually true for inputs that change at
|
||||
/// protocol rates.
|
||||
/// </param>
|
||||
/// <param name="TimerInterval">
|
||||
/// Optional periodic re-evaluation cadence. Null = timer-driven disabled. Both can
|
||||
/// be enabled simultaneously; independent scheduling paths both end at
|
||||
/// <see cref="VirtualTagEngine"/>'s <c>EvaluateInternalAsync</c>.
|
||||
/// </param>
|
||||
/// <param name="Historize">
|
||||
/// When true, every evaluation result is forwarded to the configured
|
||||
/// <see cref="IHistoryWriter"/>. Operator-set per tag; the Admin UI exposes as a
|
||||
/// checkbox.
|
||||
/// </param>
|
||||
public sealed record VirtualTagDefinition(
|
||||
string Path,
|
||||
DriverDataType DataType,
|
||||
string ScriptSource,
|
||||
bool ChangeTriggered = true,
|
||||
TimeSpan? TimerInterval = null,
|
||||
bool Historize = false);
|
||||
@@ -1,568 +0,0 @@
|
||||
using System.Collections.Concurrent;
|
||||
using Serilog;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Scripting;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
||||
|
||||
/// <summary>
|
||||
/// The Phase 7 virtual-tag evaluation engine. Ingests a set of
|
||||
/// <see cref="VirtualTagDefinition"/>s at load time, compiles each script against
|
||||
/// <see cref="ScriptSandbox"/>, builds the dependency graph, subscribes to every
|
||||
/// referenced upstream tag, and schedules re-evaluations on change + on timer.
|
||||
/// </summary>
|
||||
public sealed class VirtualTagEngine : IDisposable
|
||||
{
|
||||
private readonly ITagUpstreamSource _upstream;
|
||||
private readonly IHistoryWriter _history;
|
||||
private readonly ScriptLoggerFactory _loggerFactory;
|
||||
private readonly ILogger _engineLogger;
|
||||
private readonly Func<DateTime> _clock;
|
||||
private readonly TimeSpan _scriptTimeout;
|
||||
|
||||
private readonly DependencyGraph _graph = new();
|
||||
private readonly Dictionary<string, VirtualTagState> _tags = new(StringComparer.Ordinal);
|
||||
|
||||
/// <summary>
|
||||
/// Compile cache for every virtual-tag script. Routes <see cref="Load"/>'s
|
||||
/// <see cref="ScriptEvaluator{TContext, TResult}.Compile"/> calls through the
|
||||
/// cache so the collectible <see cref="System.Runtime.Loader.AssemblyLoadContext"/>
|
||||
/// each compile produces is actually disposed on the publish-replace path: the
|
||||
/// cache's <see cref="CompiledScriptCache{TContext, TResult}.Clear"/> disposes
|
||||
/// every materialised evaluator before dropping its dictionary entry, so a
|
||||
/// config-publish releases the prior generation's ALCs and the per-publish
|
||||
/// accretion is actually freed in production. Pre-fix, the engine called
|
||||
/// <c>ScriptEvaluator.Compile</c> directly, which left the ALCs rooted until
|
||||
/// the process exited.
|
||||
/// </summary>
|
||||
private readonly CompiledScriptCache<VirtualTagContext, object?> _compileCache = new();
|
||||
|
||||
private readonly ConcurrentDictionary<string, DataValueSnapshot> _valueCache = new(StringComparer.Ordinal);
|
||||
private readonly ConcurrentDictionary<string, List<Action<string, DataValueSnapshot>>> _observers
|
||||
= new(StringComparer.Ordinal);
|
||||
private readonly List<IDisposable> _upstreamSubscriptions = [];
|
||||
private readonly SemaphoreSlim _evalGate = new(1, 1);
|
||||
private bool _loaded;
|
||||
private bool _disposed;
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="VirtualTagEngine"/> class.
|
||||
/// </summary>
|
||||
/// <param name="upstream">The upstream tag source for reading input tags.</param>
|
||||
/// <param name="loggerFactory">Factory for creating script loggers.</param>
|
||||
/// <param name="engineLogger">Logger for engine events and errors.</param>
|
||||
/// <param name="historyWriter">Optional history writer for recording tag evaluations; defaults to no-op.</param>
|
||||
/// <param name="clock">Optional function providing current UTC time; defaults to <see cref="DateTime.UtcNow"/>.</param>
|
||||
/// <param name="scriptTimeout">Optional timeout for script evaluation; defaults to <see cref="TimedScriptEvaluator{TContext, TResult}.DefaultTimeout"/>.</param>
|
||||
public VirtualTagEngine(
|
||||
ITagUpstreamSource upstream,
|
||||
ScriptLoggerFactory loggerFactory,
|
||||
ILogger engineLogger,
|
||||
IHistoryWriter? historyWriter = null,
|
||||
Func<DateTime>? clock = null,
|
||||
TimeSpan? scriptTimeout = null)
|
||||
{
|
||||
_upstream = upstream ?? throw new ArgumentNullException(nameof(upstream));
|
||||
_loggerFactory = loggerFactory ?? throw new ArgumentNullException(nameof(loggerFactory));
|
||||
_engineLogger = engineLogger ?? throw new ArgumentNullException(nameof(engineLogger));
|
||||
_history = historyWriter ?? NullHistoryWriter.Instance;
|
||||
_clock = clock ?? (() => DateTime.UtcNow);
|
||||
_scriptTimeout = scriptTimeout ?? TimedScriptEvaluator<VirtualTagContext, object?>.DefaultTimeout;
|
||||
}
|
||||
|
||||
/// <summary>Registered tag paths, in topological order. Empty before <see cref="Load"/>.</summary>
|
||||
public IReadOnlyCollection<string> LoadedTagPaths => _tags.Keys;
|
||||
|
||||
/// <summary>Compile + register every tag in <paramref name="definitions"/>. Throws on cycle or any compile failure.</summary>
|
||||
/// <param name="definitions">List of virtual tag definitions to load and compile.</param>
|
||||
public void Load(IReadOnlyList<VirtualTagDefinition> definitions)
|
||||
{
|
||||
if (_disposed) throw new ObjectDisposedException(nameof(VirtualTagEngine));
|
||||
if (definitions is null) throw new ArgumentNullException(nameof(definitions));
|
||||
|
||||
// Start from a clean slate — supports config-publish reloads.
|
||||
UnsubscribeFromUpstream();
|
||||
_tags.Clear();
|
||||
_graph.Clear();
|
||||
_compileCache.Clear();
|
||||
|
||||
var compileFailures = new List<string>();
|
||||
var seenPaths = new HashSet<string>(StringComparer.Ordinal);
|
||||
foreach (var def in definitions)
|
||||
{
|
||||
if (!seenPaths.Add(def.Path))
|
||||
{
|
||||
compileFailures.Add($"{def.Path}: duplicate path — only one definition per path is allowed");
|
||||
continue;
|
||||
}
|
||||
|
||||
if (!IsSupportedDataType(def.DataType))
|
||||
{
|
||||
compileFailures.Add(
|
||||
$"{def.Path}: unsupported DataType DriverDataType.{def.DataType} — virtual tags only support scalar primitive types");
|
||||
continue;
|
||||
}
|
||||
|
||||
try
|
||||
{
|
||||
var extraction = DependencyExtractor.Extract(def.ScriptSource);
|
||||
if (!extraction.IsValid)
|
||||
{
|
||||
var msgs = string.Join("; ", extraction.Rejections.Select(r => r.Message));
|
||||
compileFailures.Add($"{def.Path}: dependency extraction rejected — {msgs}");
|
||||
continue;
|
||||
}
|
||||
|
||||
var evaluator = _compileCache.GetOrCompile(def.ScriptSource);
|
||||
var timed = new TimedScriptEvaluator<VirtualTagContext, object?>(evaluator, _scriptTimeout);
|
||||
var scriptLogger = _loggerFactory.Create(def.Path);
|
||||
|
||||
_tags[def.Path] = new VirtualTagState(def, timed, extraction.Reads, extraction.Writes, scriptLogger);
|
||||
_graph.Add(def.Path, extraction.Reads);
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
compileFailures.Add($"{def.Path}: {ex.Message}");
|
||||
}
|
||||
}
|
||||
|
||||
// Validate every ctx.SetVirtualTag write target resolves to a registered virtual
|
||||
// tag. A script writing to a non-existent virtual path would otherwise be silently
|
||||
// dropped at runtime by OnScriptSetVirtualTag's warning-and-drop branch; catching
|
||||
// it here surfaces operator typos as a publish failure.
|
||||
foreach (var (path, state) in _tags)
|
||||
{
|
||||
foreach (var writeTarget in state.Writes)
|
||||
{
|
||||
if (!_tags.ContainsKey(writeTarget))
|
||||
{
|
||||
compileFailures.Add(
|
||||
$"{path}: ctx.SetVirtualTag target '{writeTarget}' is not a registered virtual tag");
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
if (compileFailures.Count > 0)
|
||||
{
|
||||
var joined = string.Join("\n ", compileFailures);
|
||||
throw new InvalidOperationException(
|
||||
$"Virtual-tag engine load failed. {compileFailures.Count} script(s) did not compile:\n {joined}");
|
||||
}
|
||||
|
||||
// Cycle check — throws DependencyCycleException on offense.
|
||||
_ = _graph.TopologicalSort();
|
||||
|
||||
// Subscribe to every referenced upstream path (driver tags only — virtual tags
|
||||
// cascade internally). Seed the cache with current upstream values so first
|
||||
// evaluations see something real. Iterate _tags.Values (the registered set) rather
|
||||
// than definitions to avoid indexing by a raw input list that may contain duplicates.
|
||||
var upstreamPaths = _tags.Values
|
||||
.SelectMany(s => s.Reads)
|
||||
.Where(p => !_tags.ContainsKey(p))
|
||||
.Distinct(StringComparer.Ordinal);
|
||||
foreach (var path in upstreamPaths)
|
||||
{
|
||||
_valueCache[path] = _upstream.ReadTag(path);
|
||||
_upstreamSubscriptions.Add(_upstream.SubscribeTag(path, OnUpstreamChange));
|
||||
}
|
||||
|
||||
_loaded = true;
|
||||
_engineLogger.Information(
|
||||
"VirtualTagEngine loaded {TagCount} tag(s), {UpstreamCount} upstream subscription(s)",
|
||||
_tags.Count, _upstreamSubscriptions.Count);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Evaluate every registered tag once in topological order — used at startup so
|
||||
/// virtual tags have a defined initial value rather than inheriting the cache
|
||||
/// default. Also called after a config reload.
|
||||
/// </summary>
|
||||
/// <param name="ct">Cancellation token to stop evaluation.</param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
public async Task EvaluateAllAsync(CancellationToken ct = default)
|
||||
{
|
||||
EnsureLoaded();
|
||||
var order = _graph.TopologicalSort();
|
||||
foreach (var path in order)
|
||||
{
|
||||
if (_tags.ContainsKey(path))
|
||||
await EvaluateOneAsync(path, ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>Evaluate a single tag — used by the timer trigger + test hooks.</summary>
|
||||
/// <param name="path">Path of the virtual tag to evaluate.</param>
|
||||
/// <param name="ct">Cancellation token to stop evaluation.</param>
|
||||
/// <returns>A task that represents the asynchronous operation.</returns>
|
||||
public Task EvaluateOneAsync(string path, CancellationToken ct = default)
|
||||
{
|
||||
EnsureLoaded();
|
||||
if (!_tags.ContainsKey(path))
|
||||
throw new ArgumentException($"Not a registered virtual tag: {path}", nameof(path));
|
||||
return EvaluateInternalAsync(path, ct);
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Read the most recently evaluated value for <paramref name="path"/>. Driver
|
||||
/// tags return the last-known upstream value; virtual tags return their last
|
||||
/// evaluation result.
|
||||
/// </summary>
|
||||
/// <param name="path">Path of the tag to read.</param>
|
||||
/// <returns>The tag's most recently evaluated value, or a Bad-quality snapshot if the path is unrecognized.</returns>
|
||||
public DataValueSnapshot Read(string path)
|
||||
{
|
||||
if (string.IsNullOrWhiteSpace(path))
|
||||
return new DataValueSnapshot(null, 0x80340000u, null, _clock());
|
||||
return _valueCache.TryGetValue(path, out var v)
|
||||
? v
|
||||
: new DataValueSnapshot(null, 0x80340000u /* BadNodeIdUnknown */, null, _clock());
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Register an observer that fires on every evaluation of the given tag.
|
||||
/// Returns an <see cref="IDisposable"/> to unsubscribe. Does NOT fire a seed
|
||||
/// value — subscribers call <see cref="Read"/> for the current value if needed.
|
||||
/// </summary>
|
||||
/// <param name="path">Path of the tag to subscribe to.</param>
|
||||
/// <param name="observer">Callback invoked with the tag path and new value on each evaluation.</param>
|
||||
/// <returns>An <see cref="IDisposable"/> that removes the observer when disposed.</returns>
|
||||
public IDisposable Subscribe(string path, Action<string, DataValueSnapshot> observer)
|
||||
{
|
||||
// Race-safe pattern paired with Unsub.Dispose: if Unsub.Dispose removed the
|
||||
// dictionary entry between our GetOrAdd and the lock-protected Add, the list
|
||||
// we hold a reference to is orphaned. Re-check the map under the lock and
|
||||
// re-insert the list (or grab the current one) if needed, retrying until the
|
||||
// dictionary observably contains the list we just added our observer to.
|
||||
while (true)
|
||||
{
|
||||
var list = _observers.GetOrAdd(path, _ => []);
|
||||
lock (list)
|
||||
{
|
||||
// Confirm the list is still the dictionary's value for this key. If
|
||||
// Dispose removed the entry, _observers[path] either doesn't exist or
|
||||
// points at a different (newer) list — retry.
|
||||
if (_observers.TryGetValue(path, out var current) && ReferenceEquals(current, list))
|
||||
{
|
||||
list.Add(observer);
|
||||
return new Unsub(this, path, observer);
|
||||
}
|
||||
}
|
||||
// Lost the race — Dispose pruned the list out from under us. Loop and
|
||||
// either re-create or pick up the newer list.
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Change-trigger entry point — called by the upstream subscription callback.
|
||||
/// Updates the cache, fans out to observers (so OPC UA clients see the upstream
|
||||
/// change too if they subscribed via the engine), and schedules every
|
||||
/// change-triggered dependent for re-evaluation in topological order.
|
||||
/// </summary>
|
||||
/// <param name="path">Path of the upstream tag that changed.</param>
|
||||
/// <param name="value">New value of the tag.</param>
|
||||
internal void OnUpstreamChange(string path, DataValueSnapshot value)
|
||||
{
|
||||
_valueCache[path] = value;
|
||||
NotifyObservers(path, value);
|
||||
|
||||
// Fire-and-forget — the upstream subscription callback must not block the
|
||||
// driver's dispatcher. Exceptions during cascade are handled per-tag inside
|
||||
// EvaluateInternalAsync.
|
||||
_ = CascadeAsync(path, CancellationToken.None);
|
||||
}
|
||||
|
||||
private async Task CascadeAsync(string upstreamPath, CancellationToken ct)
|
||||
{
|
||||
try
|
||||
{
|
||||
var dependents = _graph.TransitiveDependentsInOrder(upstreamPath);
|
||||
foreach (var dep in dependents)
|
||||
{
|
||||
if (_tags.TryGetValue(dep, out var state) && state.Definition.ChangeTriggered)
|
||||
await EvaluateInternalAsync(dep, ct).ConfigureAwait(false);
|
||||
}
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
_engineLogger.Error(ex, "VirtualTagEngine cascade failed for upstream {Path}", upstreamPath);
|
||||
}
|
||||
}
|
||||
|
||||
private async Task EvaluateInternalAsync(string path, CancellationToken ct)
|
||||
{
|
||||
if (!_tags.TryGetValue(path, out var state)) return;
|
||||
|
||||
await _evalGate.WaitAsync(ct).ConfigureAwait(false);
|
||||
try
|
||||
{
|
||||
var ctxCache = BuildReadCache(state.Reads);
|
||||
|
||||
// Cold-start guard — when any upstream input is still unset or Bad-quality,
|
||||
// publish a BadWaitingForInitialData snapshot so OPC UA clients see a defined
|
||||
// quality rather than observing "not yet computed" as a stale Good value.
|
||||
// Evaluating with nulls would throw inside the script (scripts cast
|
||||
// ctx.GetTag(path).Value directly) and produce a persistent BadInternalError.
|
||||
if (!AreInputsReady(ctxCache))
|
||||
{
|
||||
var notReady = new DataValueSnapshot(null, 0x80320000u /* BadWaitingForInitialData */, null, _clock());
|
||||
_valueCache[path] = notReady;
|
||||
NotifyObservers(path, notReady);
|
||||
return;
|
||||
}
|
||||
|
||||
var context = new VirtualTagContext(
|
||||
ctxCache,
|
||||
(p, v) => OnScriptSetVirtualTag(p, v),
|
||||
state.Logger,
|
||||
_clock);
|
||||
|
||||
DataValueSnapshot result;
|
||||
try
|
||||
{
|
||||
var raw = await state.Evaluator.RunAsync(context, ct).ConfigureAwait(false);
|
||||
var coerced = CoerceResult(raw, state.Definition.DataType);
|
||||
// null from CoerceResult means the conversion threw (raw was non-null but
|
||||
// not convertible to the declared type). Surface as BadInternalError so
|
||||
// the OPC UA client sees a defined Bad quality rather than a Good null.
|
||||
result = (raw is not null && coerced is null)
|
||||
? new DataValueSnapshot(null, 0x80020000u /* BadInternalError */, null, _clock())
|
||||
: new DataValueSnapshot(coerced, 0u, _clock(), _clock());
|
||||
}
|
||||
catch (ScriptTimeoutException tex)
|
||||
{
|
||||
state.Logger.Warning("Script timed out after {Timeout}", tex.Timeout);
|
||||
result = new DataValueSnapshot(null, 0x80020000u /* BadInternalError */, null, _clock());
|
||||
}
|
||||
catch (OperationCanceledException)
|
||||
{
|
||||
throw; // shutdown path — don't misclassify
|
||||
}
|
||||
catch (Exception ex)
|
||||
{
|
||||
state.Logger.Error(ex, "Virtual-tag script threw");
|
||||
result = new DataValueSnapshot(null, 0x80020000u /* BadInternalError */, null, _clock());
|
||||
}
|
||||
|
||||
_valueCache[path] = result;
|
||||
NotifyObservers(path, result);
|
||||
if (state.Definition.Historize) _history.Record(path, result);
|
||||
}
|
||||
finally
|
||||
{
|
||||
_evalGate.Release();
|
||||
}
|
||||
}
|
||||
|
||||
private IReadOnlyDictionary<string, DataValueSnapshot> BuildReadCache(IReadOnlySet<string> reads)
|
||||
{
|
||||
var map = new Dictionary<string, DataValueSnapshot>(StringComparer.Ordinal);
|
||||
foreach (var r in reads)
|
||||
{
|
||||
map[r] = _valueCache.TryGetValue(r, out var v)
|
||||
? v
|
||||
: _upstream.ReadTag(r);
|
||||
}
|
||||
return map;
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Returns true when every entry in <paramref name="cache"/> has a
|
||||
/// Good/Uncertain-quality <see cref="DataValueSnapshot.StatusCode"/> (bit 31 = 0).
|
||||
/// Readiness is gated on <c>StatusCode</c> alone — a Good-quality snapshot with a
|
||||
/// null <c>Value</c> (e.g. a script that returns <c>null</c>, or a
|
||||
/// <c>ctx.SetVirtualTag</c> call with a null argument) is considered ready and the
|
||||
/// downstream script is allowed to run. The downstream script will observe the null
|
||||
/// value and typically throw an NRE, which the outer evaluation catch maps to
|
||||
/// <c>BadInternalError</c> — the correct sentinel for "script ran but faulted"
|
||||
/// rather than "inputs not yet available". Previously the null-value check
|
||||
/// caused a Good-quality null upstream to permanently block all its dependents at
|
||||
/// <c>BadWaitingForInitialData</c>.
|
||||
/// </summary>
|
||||
private static bool AreInputsReady(IReadOnlyDictionary<string, DataValueSnapshot> cache)
|
||||
{
|
||||
foreach (var kv in cache)
|
||||
{
|
||||
if (kv.Value is null) return false;
|
||||
if ((kv.Value.StatusCode & 0x80000000u) != 0) return false;
|
||||
}
|
||||
return true;
|
||||
}
|
||||
|
||||
private void OnScriptSetVirtualTag(string path, object? value)
|
||||
{
|
||||
if (!_tags.ContainsKey(path))
|
||||
{
|
||||
_engineLogger.Warning(
|
||||
"Script attempted ctx.SetVirtualTag on non-virtual or non-registered path {Path}", path);
|
||||
return;
|
||||
}
|
||||
var snap = new DataValueSnapshot(value, 0u, _clock(), _clock());
|
||||
_valueCache[path] = snap;
|
||||
NotifyObservers(path, snap);
|
||||
if (_tags[path].Definition.Historize) _history.Record(path, snap);
|
||||
|
||||
// A cross-tag write must participate in the change-trigger cascade, exactly
|
||||
// like an upstream delta — any change-triggered tag that reads this path
|
||||
// would otherwise go stale until an unrelated trigger fires (see
|
||||
// docs/VirtualTags.md, VirtualTagContext section). Fire-and-forget: this
|
||||
// callback runs inside EvaluateInternalAsync with the non-reentrant
|
||||
// _evalGate held, so the cascade must be scheduled, not invoked inline.
|
||||
_ = CascadeAsync(path, CancellationToken.None);
|
||||
}
|
||||
|
||||
private void NotifyObservers(string path, DataValueSnapshot value)
|
||||
{
|
||||
if (!_observers.TryGetValue(path, out var list)) return;
|
||||
Action<string, DataValueSnapshot>[] snapshot;
|
||||
lock (list) { snapshot = list.ToArray(); }
|
||||
foreach (var obs in snapshot)
|
||||
{
|
||||
try { obs(path, value); }
|
||||
catch (Exception ex)
|
||||
{
|
||||
_engineLogger.Warning(ex, "Virtual-tag observer for {Path} threw", path);
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
private static object? CoerceResult(object? raw, DriverDataType target)
|
||||
{
|
||||
if (raw is null) return null;
|
||||
try
|
||||
{
|
||||
return target switch
|
||||
{
|
||||
DriverDataType.Boolean => Convert.ToBoolean(raw),
|
||||
DriverDataType.Int16 => Convert.ToInt16(raw),
|
||||
DriverDataType.Int32 => Convert.ToInt32(raw),
|
||||
DriverDataType.Int64 => Convert.ToInt64(raw),
|
||||
DriverDataType.UInt16 => Convert.ToUInt16(raw),
|
||||
DriverDataType.UInt32 => Convert.ToUInt32(raw),
|
||||
DriverDataType.UInt64 => Convert.ToUInt64(raw),
|
||||
DriverDataType.Float32 => Convert.ToSingle(raw),
|
||||
DriverDataType.Float64 => Convert.ToDouble(raw),
|
||||
DriverDataType.String => Convert.ToString(raw) ?? string.Empty,
|
||||
DriverDataType.DateTime => raw is DateTime dt ? dt : Convert.ToDateTime(raw),
|
||||
// Any DriverDataType not in the explicit list (currently Reference, or any
|
||||
// future enum member added without coercion support) must NOT silently
|
||||
// return the uncoerced raw value — that would surface as a wire-level
|
||||
// type mismatch on the OPC UA Variant. Throwing here is caught by the
|
||||
// outer catch and mapped to BadInternalError. Load-time validation in
|
||||
// IsSupportedDataType ensures operators never publish such a tag.
|
||||
_ => throw new InvalidOperationException(
|
||||
$"Virtual-tag CoerceResult does not support DriverDataType.{target}"),
|
||||
};
|
||||
}
|
||||
catch
|
||||
{
|
||||
// Caller maps null to BadInternalError (0x80020000) — we let null propagate
|
||||
// so the outer evaluation path sets the Bad quality on the snapshot.
|
||||
return null;
|
||||
}
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// The set of <see cref="DriverDataType"/> values <see cref="CoerceResult"/> can
|
||||
/// honour. Definitions declaring any other type are rejected at <see cref="Load"/>
|
||||
/// so an operator typo (or a future enum member added without coercion support) is
|
||||
/// caught at publish time rather than silently producing a type-mismatched value.
|
||||
/// </summary>
|
||||
private static bool IsSupportedDataType(DriverDataType t) => t switch
|
||||
{
|
||||
DriverDataType.Boolean => true,
|
||||
DriverDataType.Int16 => true,
|
||||
DriverDataType.Int32 => true,
|
||||
DriverDataType.Int64 => true,
|
||||
DriverDataType.UInt16 => true,
|
||||
DriverDataType.UInt32 => true,
|
||||
DriverDataType.UInt64 => true,
|
||||
DriverDataType.Float32 => true,
|
||||
DriverDataType.Float64 => true,
|
||||
DriverDataType.String => true,
|
||||
DriverDataType.DateTime => true,
|
||||
_ => false,
|
||||
};
|
||||
|
||||
private void UnsubscribeFromUpstream()
|
||||
{
|
||||
foreach (var s in _upstreamSubscriptions)
|
||||
{
|
||||
try { s.Dispose(); } catch { /* best effort */ }
|
||||
}
|
||||
_upstreamSubscriptions.Clear();
|
||||
}
|
||||
|
||||
private void EnsureLoaded()
|
||||
{
|
||||
if (!_loaded) throw new InvalidOperationException(
|
||||
"VirtualTagEngine not loaded. Call Load(definitions) first.");
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Releases all resources held by the engine, including upstream subscriptions and compiled script caches.
|
||||
/// </summary>
|
||||
public void Dispose()
|
||||
{
|
||||
if (_disposed) return;
|
||||
_disposed = true;
|
||||
UnsubscribeFromUpstream();
|
||||
_tags.Clear();
|
||||
_graph.Clear();
|
||||
_compileCache.Dispose();
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Gets the dependency graph for testing purposes.
|
||||
/// </summary>
|
||||
internal DependencyGraph GraphForTesting => _graph;
|
||||
|
||||
private sealed class Unsub : IDisposable
|
||||
{
|
||||
private readonly VirtualTagEngine _engine;
|
||||
private readonly string _path;
|
||||
private readonly Action<string, DataValueSnapshot> _observer;
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="Unsub"/> class.
|
||||
/// </summary>
|
||||
/// <param name="e">The parent virtual tag engine.</param>
|
||||
/// <param name="path">Path of the subscribed tag.</param>
|
||||
/// <param name="observer">The observer callback to unsubscribe.</param>
|
||||
public Unsub(VirtualTagEngine e, string path, Action<string, DataValueSnapshot> observer)
|
||||
{
|
||||
_engine = e; _path = path; _observer = observer;
|
||||
}
|
||||
/// <summary>
|
||||
/// Removes the observer from the subscription list.
|
||||
/// </summary>
|
||||
public void Dispose()
|
||||
{
|
||||
if (_engine._observers.TryGetValue(_path, out var list))
|
||||
{
|
||||
lock (list)
|
||||
{
|
||||
list.Remove(_observer);
|
||||
// If we removed the last observer, prune the dictionary entry so a
|
||||
// long-running server doesn't accumulate empty Lists for paths that
|
||||
// saw transient subscriptions. The emptiness check is inside the same
|
||||
// lock so a concurrent Subscribe can't slip an observer in after we
|
||||
// observe the list as empty.
|
||||
if (list.Count == 0)
|
||||
{
|
||||
// ICollection<KeyValuePair<,>> removal is value-typed — only removes
|
||||
// if both key + value still match (i.e. the dictionary still points
|
||||
// at this list). This keeps a racing Subscribe's brand-new list safe.
|
||||
((ICollection<KeyValuePair<string, List<Action<string, DataValueSnapshot>>>>)_engine._observers)
|
||||
.Remove(new KeyValuePair<string, List<Action<string, DataValueSnapshot>>>(_path, list));
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
}
|
||||
|
||||
internal sealed record VirtualTagState(
|
||||
VirtualTagDefinition Definition,
|
||||
TimedScriptEvaluator<VirtualTagContext, object?> Evaluator,
|
||||
IReadOnlySet<string> Reads,
|
||||
IReadOnlySet<string> Writes,
|
||||
ILogger Logger);
|
||||
}
|
||||
@@ -1,106 +0,0 @@
|
||||
using System.Collections.Concurrent;
|
||||
using ZB.MOM.WW.OtOpcUa.Core.Abstractions;
|
||||
|
||||
namespace ZB.MOM.WW.OtOpcUa.Core.VirtualTags;
|
||||
|
||||
/// <summary>
|
||||
/// Implements the driver-agnostic capability surface the
|
||||
/// <c>DriverNodeManager</c> dispatches to when a node resolves to
|
||||
/// <c>NodeSource.Virtual</c>. Reads return the engine's last-known
|
||||
/// evaluation result; subscriptions forward engine-emitted change events as
|
||||
/// <see cref="ISubscribable.OnDataChange"/> events.
|
||||
/// </summary>
|
||||
/// <remarks>
|
||||
/// <para>
|
||||
/// <see cref="IWritable"/> is deliberately not implemented — OPC UA client
|
||||
/// writes to virtual tags are rejected in <c>DriverNodeManager</c> before they
|
||||
/// reach here. Scripts are the only write path, routed
|
||||
/// through <c>ctx.SetVirtualTag</c>.
|
||||
/// </para>
|
||||
/// </remarks>
|
||||
public sealed class VirtualTagSource : IReadable, ISubscribable
|
||||
{
|
||||
private readonly VirtualTagEngine _engine;
|
||||
private readonly ConcurrentDictionary<string, Subscription> _subs = new(StringComparer.Ordinal);
|
||||
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="VirtualTagSource"/> class.
|
||||
/// </summary>
|
||||
/// <param name="engine">The virtual tag engine.</param>
|
||||
public VirtualTagSource(VirtualTagEngine engine)
|
||||
{
|
||||
_engine = engine ?? throw new ArgumentNullException(nameof(engine));
|
||||
}
|
||||
|
||||
/// <summary>
|
||||
/// Occurs when data values change.
|
||||
/// </summary>
|
||||
public event EventHandler<DataChangeEventArgs>? OnDataChange;
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<IReadOnlyList<DataValueSnapshot>> ReadAsync(
|
||||
IReadOnlyList<string> fullReferences, CancellationToken cancellationToken)
|
||||
{
|
||||
if (fullReferences is null) throw new ArgumentNullException(nameof(fullReferences));
|
||||
var results = new DataValueSnapshot[fullReferences.Count];
|
||||
for (var i = 0; i < fullReferences.Count; i++)
|
||||
results[i] = _engine.Read(fullReferences[i]);
|
||||
return Task.FromResult<IReadOnlyList<DataValueSnapshot>>(results);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task<ISubscriptionHandle> SubscribeAsync(
|
||||
IReadOnlyList<string> fullReferences,
|
||||
TimeSpan publishingInterval,
|
||||
CancellationToken cancellationToken)
|
||||
{
|
||||
if (fullReferences is null) throw new ArgumentNullException(nameof(fullReferences));
|
||||
|
||||
var handle = new SubscriptionHandle(Guid.NewGuid().ToString("N"));
|
||||
var observers = new List<IDisposable>(fullReferences.Count);
|
||||
|
||||
// OPC UA convention: for each path, emit the initial-data callback BEFORE
|
||||
// registering the change observer. This prevents a race where an upstream change
|
||||
// fires the observer between the Subscribe call and the Read call, which would
|
||||
// deliver a newer change event before the initial-data event, leaving the client
|
||||
// with a stale last-known value.
|
||||
foreach (var path in fullReferences)
|
||||
{
|
||||
var snap = _engine.Read(path);
|
||||
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, path, snap));
|
||||
observers.Add(_engine.Subscribe(path, (p, s) =>
|
||||
OnDataChange?.Invoke(this, new DataChangeEventArgs(handle, p, s))));
|
||||
}
|
||||
_subs[handle.DiagnosticId] = new Subscription(handle, observers);
|
||||
|
||||
return Task.FromResult<ISubscriptionHandle>(handle);
|
||||
}
|
||||
|
||||
/// <inheritdoc />
|
||||
public Task UnsubscribeAsync(ISubscriptionHandle handle, CancellationToken cancellationToken)
|
||||
{
|
||||
if (handle is null) throw new ArgumentNullException(nameof(handle));
|
||||
if (_subs.TryRemove(handle.DiagnosticId, out var sub))
|
||||
{
|
||||
foreach (var d in sub.Observers)
|
||||
{
|
||||
try { d.Dispose(); } catch { }
|
||||
}
|
||||
}
|
||||
return Task.CompletedTask;
|
||||
}
|
||||
|
||||
private sealed class SubscriptionHandle : ISubscriptionHandle
|
||||
{
|
||||
/// <summary>
|
||||
/// Initializes a new instance of the <see cref="SubscriptionHandle"/> class.
|
||||
/// </summary>
|
||||
/// <param name="id">The subscription identifier.</param>
|
||||
public SubscriptionHandle(string id) { DiagnosticId = id; }
|
||||
|
||||
/// <inheritdoc />
|
||||
public string DiagnosticId { get; }
|
||||
}
|
||||
|
||||
private sealed record Subscription(SubscriptionHandle Handle, IReadOnlyList<IDisposable> Observers);
|
||||
}
|
||||
@@ -3,12 +3,10 @@ using Microsoft.Extensions.Logging;
|
||||
namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation;
|
||||
|
||||
/// <summary>
|
||||
/// Periodic re-evaluation scheduler for calc tags carrying a <c>timerIntervalMs</c>. A deliberate
|
||||
/// mirror of the VirtualTag <c>TimerTriggerScheduler</c> (<c>Core.VirtualTags</c>): one
|
||||
/// Periodic re-evaluation scheduler for calc tags carrying a <c>timerIntervalMs</c>. One
|
||||
/// <see cref="System.Threading.Timer"/> per distinct interval group (so the wire count stays low
|
||||
/// regardless of tag count) with a per-group in-flight flag that skips a tick when the prior tick for
|
||||
/// the same group is still running. Revived <b>inside the driver</b> because the VT scheduler is welded
|
||||
/// to <c>VirtualTagEngine.EvaluateOneAsync</c>; here each tick invokes the driver's own evaluate
|
||||
/// the same group is still running. Each tick invokes the driver's own evaluate
|
||||
/// callback per tag in the group.
|
||||
/// </summary>
|
||||
internal sealed class CalculationTimerScheduler : IDisposable
|
||||
|
||||
@@ -22,7 +22,7 @@ public sealed class LastValueCache
|
||||
{
|
||||
// OPC UA BadWaitingForInitialData (0x80320000) — the repo-wide convention for "no value has
|
||||
// been observed yet" (mirrored by CalculationDriver before its first evaluation,
|
||||
// VirtualTagEngine for a freshly-materialised node, and OtOpcUaNodeManager /
|
||||
// and OtOpcUaNodeManager /
|
||||
// AddressSpaceApplier for a just-deployed variable). MQTT is subscribe-first, so an unseen
|
||||
// RawPath is in exactly that state until its first publish arrives. Deliberately NOT
|
||||
// GoodNoData — that code is reserved (see NullHistorianDataSource, OtOpcUaNodeManager
|
||||
|
||||
@@ -347,7 +347,7 @@ public sealed class ScriptAnalysisService
|
||||
: $"**Tag path** `{Code(info.Path)}`\n\n{info.Kind} · Type **{info.DataType}**"
|
||||
+ (info.DriverInstanceId is null ? "" : $" · Driver `{Code(info.DriverInstanceId)}`");
|
||||
// Truthfulness: ctx.SetVirtualTag(...) is a no-op in the live single-tag evaluator
|
||||
// (RoslynVirtualTagEvaluator drops cross-tag writes; the cascade VirtualTagEngine is dormant).
|
||||
// (RoslynVirtualTagEvaluator drops cross-tag writes).
|
||||
if (isSetVirtualTag)
|
||||
tagMd += "\n\n⚠ Cross-tag `SetVirtualTag` writes are currently dropped in single-tag mode " +
|
||||
"(the cascade engine is not wired into the host); this write will not take effect at runtime.";
|
||||
|
||||
@@ -16,7 +16,7 @@ namespace ZB.MOM.WW.OtOpcUa.Host.Engines;
|
||||
///
|
||||
/// Single-tag mode: cross-tag <c>ctx.SetVirtualTag</c> writes are dropped (logged) because
|
||||
/// fan-out between actors is owned by <c>DependencyMuxActor</c>, not by the eval engine.
|
||||
/// Cycle detection + cascade ordering live in <see cref="VirtualTagEngine"/>; this adapter
|
||||
/// Cycle detection lives in the deploy gate (<c>DependencyGraph</c>); this adapter
|
||||
/// stays single-tag scoped to keep <see cref="VirtualTagActor"/>'s message loop simple.
|
||||
///
|
||||
/// <para>
|
||||
|
||||
@@ -366,7 +366,7 @@ public sealed class VirtualTagActor : ReceiveActor
|
||||
_virtualTagId, _hasLastValue ? _lastValue : null, timestampUtc, CorrelationId.NewId(), OpcUaStatus.BadInternalError));
|
||||
}
|
||||
|
||||
/// <summary>Inputs-ready gate (mirrors the dormant <c>VirtualTagEngine.AreInputsReady</c>): true once
|
||||
/// <summary>Inputs-ready gate (the retired cascade engine's rule): true once
|
||||
/// every declared dependency ref has been seen at least once. Called once per failure — the ref list
|
||||
/// is small, so no caching. A vacuously-true empty ref list means a no-dependency script degrades on
|
||||
/// its first failure.</summary>
|
||||
|
||||
Reference in New Issue
Block a user