using Microsoft.Extensions.Logging;
namespace ZB.MOM.WW.OtOpcUa.Driver.Calculation;
///
/// Periodic re-evaluation scheduler for calc tags carrying a timerIntervalMs. One
/// 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. Each tick invokes the driver's own evaluate
/// callback per tag in the group and AWAITS it — the in-flight flag is held across the awaited script
/// run, not just across the synchronous part of the callback, so a slow script still bounds its group
/// to one outstanding tick.
///
internal sealed class CalculationTimerScheduler : IDisposable
{
private readonly Func _evaluate;
private readonly ILogger _logger;
private readonly List _timers = [];
private readonly List _groups = [];
private readonly CancellationTokenSource _cts = new();
private long _skippedTickCount;
private bool _disposed;
/// Initializes a new instance of the class.
/// Callback invoked with a calc tag's RawPath when its timer group ticks; awaited.
/// Logger for diagnostics.
public CalculationTimerScheduler(Func evaluate, ILogger logger)
{
_evaluate = evaluate ?? throw new ArgumentNullException(nameof(evaluate));
_logger = logger ?? throw new ArgumentNullException(nameof(logger));
}
/// Number of timer callbacks that skipped their work because the prior tick for the same
/// group was still running. Monotonic; exposed for tests + operational metrics.
public long SkippedTickCount => Interlocked.Read(ref _skippedTickCount);
/// Stand up one per distinct interval. All tags sharing an interval share
/// a timer; each tick re-evaluates every tag in the group.
/// The calc tags to schedule (only those with a positive are scheduled).
public void Start(IEnumerable tags)
{
ArgumentNullException.ThrowIfNull(tags);
if (_disposed) throw new ObjectDisposedException(nameof(CalculationTimerScheduler));
var byInterval = tags
.Where(t => t.TimerInterval is { } iv && iv > TimeSpan.Zero)
.GroupBy(t => t.TimerInterval!.Value);
foreach (var group in byInterval)
{
var paths = group.Select(t => t.RawPath).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.LogInformation("CalculationTimerScheduler: {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 — bounds the outstanding
// work to one tick per group regardless of how long an evaluation takes.
if (Interlocked.CompareExchange(ref ctx.InFlight, 1, 0) != 0)
{
Interlocked.Increment(ref _skippedTickCount);
return;
}
// The Timer callback is void; the tick itself is async so the awaited script runs do not pin this
// pool thread. TickAsync owns the in-flight flag and swallows everything — nothing can escape to the
// Timer (which would be an unobserved fault on a timer thread).
_ = TickAsync(ctx);
}
private async Task TickAsync(TickGroup ctx)
{
try
{
foreach (var path in ctx.Paths)
{
if (_cts.IsCancellationRequested) return;
try { await _evaluate(path).ConfigureAwait(false); }
catch (Exception ex) { _logger.LogError(ex, "CalculationTimerScheduler evaluate failed for {Path}", path); }
}
}
finally
{
Interlocked.Exchange(ref ctx.InFlight, 0);
}
}
/// Releases all timers and disposes the scheduler's resources.
public void Dispose()
{
if (_disposed) return;
_disposed = true;
_cts.Cancel();
foreach (var t in _timers)
{
try { t.Dispose(); } catch { /* best-effort teardown */ }
}
_timers.Clear();
_groups.Clear();
_cts.Dispose();
}
private sealed class TickGroup(IReadOnlyList paths)
{
// 0 = idle, 1 = a tick is currently running for this group.
public int InFlight;
/// The paths evaluated on each tick of this group.
public IReadOnlyList Paths { get; } = paths;
}
}