/* ======================================================================== * Copyright (c) 2005-2025 The OPC Foundation, Inc. All rights reserved. * * OPC Foundation MIT License 1.00 * * Permission is hereby granted, free of charge, to any person * obtaining a copy of this software and associated documentation * files (the "Software"), to deal in the Software without * restriction, including without limitation the rights to use, * copy, modify, merge, publish, distribute, sublicense, and/or sell * copies of the Software, and to permit persons to whom the * Software is furnished to do so, subject to the following * conditions: * * The above copyright notice and this permission notice shall be * included in all copies or substantial portions of the Software. * THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, * EXPRESS OR IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES * OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE AND * NONINFRINGEMENT. IN NO EVENT SHALL THE AUTHORS OR COPYRIGHT * HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER LIABILITY, * WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING * FROM, OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR * OTHER DEALINGS IN THE SOFTWARE. * * The complete license agreement can be found here: * http://opcfoundation.org/License/MIT/1.00/ * ======================================================================*/ using System; using System.Threading; using System.Threading.Tasks; using Microsoft.Extensions.Logging; namespace Opc.Ua.PubSub { /// /// component that is specialized in calculating and executing a routine for a given interval /// public class IntervalRunner : IDisposable { private const int kMinInterval = 10; private readonly Lock m_lock = new(); private readonly ILogger m_logger; private readonly TimeProvider m_timeProvider; private double m_interval = kMinInterval; private double m_nextPublishTick; /// /// event used to cancel run /// private CancellationTokenSource? m_cancellationToken = new(); /// /// Create new instance of . /// public IntervalRunner( object? id, double interval, Func canExecuteFunc, Func intervalActionAsync, ITelemetryContext telemetry, TimeProvider? timeProvider = null) { m_logger = telemetry.CreateLogger(); Id = id; Interval = interval; CanExecuteFunc = canExecuteFunc; IntervalActionAsync = intervalActionAsync; m_timeProvider = timeProvider ?? TimeProvider.System; } /// /// Identifier of current IntervalRunner /// public object? Id { get; } /// /// Get/set the Interval between Runs in milliseconds /// public double Interval { get => m_interval; set { lock (m_lock) { if (value < kMinInterval) { value = kMinInterval; } m_interval = value; } } } /// /// Get the function that decides if the configured action can be executed when the Interval elapses /// public Func CanExecuteFunc { get; } /// /// Get the async action that will be executed at each interval /// public Func IntervalActionAsync { get; } /// /// Starts the IntervalRunner and makes it ready to execute the code. /// public void Start() { lock (m_lock) { // m_cancellationToken is only null after Dispose; the existing contract // assumes Start is not called after Dispose (latent NRE preserved). // TODO: Consider adding ObjectDisposedException check after Dispose. if (m_cancellationToken!.IsCancellationRequested) { m_cancellationToken.Dispose(); m_cancellationToken = new CancellationTokenSource(); } } Task.Run(ProcessAsync).ConfigureAwait(false); m_logger.LogInformation("IntervalRunner with id: {Id} was started.", Id); } /// /// Stop the publishing thread. /// public virtual void Stop() { lock (m_lock) { m_cancellationToken?.Cancel(); } m_logger.LogInformation("IntervalRunner with id: {Id} was stopped.", Id); } /// /// Releases all resources used by the current instance of the class. /// public void Dispose() { Dispose(true); GC.SuppressFinalize(this); } /// /// When overridden in a derived class, releases the unmanaged resources used by that class /// and optionally releases the managed resources. /// /// true to release both managed and unmanaged resources; false to release only unmanaged resources. protected virtual void Dispose(bool disposing) { if (disposing) { Stop(); m_cancellationToken?.Dispose(); m_cancellationToken = null; } } /// /// Periodically executes the . /// private async Task ProcessAsync() { bool isSystemProvider = ReferenceEquals(m_timeProvider, TimeProvider.System); long ticksPerMs = Math.Max(1L, m_timeProvider.TimestampFrequency / 1000L); lock (m_lock) { m_nextPublishTick = m_timeProvider.GetTimestamp(); } // m_cancellationToken is only null after Dispose; ProcessAsync is started // by Start before Dispose can be reached in normal usage. while (!m_cancellationToken!.IsCancellationRequested) { long nowTick = m_timeProvider.GetTimestamp(); double nextPublishTick = 0; lock (m_lock) { nextPublishTick = m_nextPublishTick; } double sleepCycle = (nextPublishTick - nowTick) / ticksPerMs; if (sleepCycle > 16) { // Use Task.Delay if sleep cycle is larger await m_timeProvider.Delay( TimeSpan.FromMilliseconds(sleepCycle), m_cancellationToken.Token) .ConfigureAwait(false); // Still ticks to consume (spurious wakeup too early), improbable nowTick = m_timeProvider.GetTimestamp(); if (nowTick < nextPublishTick) { if (isSystemProvider) { SpinWait.SpinUntil(() => m_timeProvider.GetTimestamp() >= nextPublishTick); } else { double remainingMs = (nextPublishTick - nowTick) / ticksPerMs; if (remainingMs > 0) { await m_timeProvider.Delay( TimeSpan.FromMilliseconds(remainingMs), m_cancellationToken.Token) .ConfigureAwait(false); } } } } else if (sleepCycle is >= 0 and <= 16) { if (isSystemProvider) { SpinWait.SpinUntil(() => m_timeProvider.GetTimestamp() >= nextPublishTick); } else if (sleepCycle > 0) { await m_timeProvider.Delay( TimeSpan.FromMilliseconds(sleepCycle), m_cancellationToken.Token) .ConfigureAwait(false); } } lock (m_lock) { double nextCycle = (long)m_interval * ticksPerMs; m_nextPublishTick += nextCycle; if (IntervalActionAsync != null && CanExecuteFunc != null && CanExecuteFunc()) { // call on a new task without blocking the thread pool _ = Task.Run(IntervalActionAsync); } } } } } }