/* ========================================================================
* 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);
}
}
}
}
}
}