/* ========================================================================
* 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.Collections.Generic;
using System.Globalization;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
namespace Opc.Ua.Server
{
///
/// An object which periodically reads the items and updates the cache.
///
public class SamplingGroup : IDisposable
{
///
/// Creates a new instance of a sampling group.
///
public SamplingGroup(
IServerInternal server,
IAsyncNodeManager nodeManager,
List samplingRates,
OperationContext context,
double samplingInterval,
IUserIdentity? savedOwnerIdentity = null)
: this(
server,
nodeManager,
samplingRates,
context,
samplingInterval,
savedOwnerIdentity,
timeProvider: null)
{
}
///
/// Creates a new instance of a sampling group with an explicit
/// .
///
/// The server.
/// The owning node manager.
/// The supported sampling-rate groups.
/// The operation context.
/// The requested sampling interval.
/// Owner identity for sessionless groups.
///
/// Optional used for the sampling-loop pacing
/// and for the wall-clock fallback timestamp emitted when a sampled
/// read returns no value. When null, the time provider exposed
/// by the server (via ) is used,
/// falling back to .
///
public SamplingGroup(
IServerInternal server,
IAsyncNodeManager nodeManager,
List samplingRates,
OperationContext context,
double samplingInterval,
IUserIdentity? savedOwnerIdentity,
TimeProvider? timeProvider)
{
m_server = server ?? throw new ArgumentNullException(nameof(server));
m_timeProvider = timeProvider
?? (server as ITimeProviderProvider)?.TimeProvider
?? TimeProvider.System;
m_logger = server.Telemetry.CreateLogger();
m_nodeManager = nodeManager ?? throw new ArgumentNullException(nameof(nodeManager));
m_samplingRates = samplingRates ??
throw new ArgumentNullException(nameof(samplingRates));
m_session = context.Session;
if (m_session == null)
{
m_effectiveIdentity =
savedOwnerIdentity
?? throw new ArgumentNullException(
nameof(savedOwnerIdentity),
"Either a context with a Session or an owner identity need to be provided");
}
m_diagnosticsMask = context.DiagnosticsMask & DiagnosticsMasks.OperationAll;
m_samplingInterval = AdjustSamplingInterval(samplingInterval);
m_itemsToAdd = [];
m_itemsToRemove = [];
m_items = [];
// create a event to signal shutdown.
m_shutdownEvent = new ManualResetEvent(true);
}
///
/// Frees any unmanaged resources.
///
public void Dispose()
{
Dispose(true);
GC.SuppressFinalize(this);
}
///
/// An overrideable version of the Dispose.
///
protected virtual void Dispose(bool disposing)
{
if (disposing)
{
lock (m_lock)
{
m_shutdownEvent.Set();
m_samplingRates.Clear();
}
if (m_samplingTask != null)
{
try
{
m_samplingTask.GetAwaiter().GetResult();
}
catch (AggregateException)
{ /* Ignore exceptions on shutdown */
}
}
m_samplingTask?.Dispose();
m_shutdownEvent.Dispose();
}
}
///
/// Starts the sampling thread which periodically reads the items in the group.
///
public void Startup()
{
lock (m_lock)
{
m_shutdownEvent.Reset();
m_samplingTask = Task.Factory.StartNew(
() => SampleMonitoredItemsAsync(m_samplingInterval, CancellationToken.None).AsTask(),
default, // TODO: Pass a cancellation token
TaskCreationOptions.LongRunning | TaskCreationOptions.DenyChildAttach,
TaskScheduler.Default).Unwrap();
}
}
///
/// Stops the sampling thread.
///
public void Shutdown()
{
lock (m_lock)
{
m_shutdownEvent.Set();
m_items.Clear();
m_samplingTask = null;
}
}
///
/// Checks if the monitored item can be handled by the group.
///
///
/// True if the item was added to the group.
///
///
/// The ApplyChanges() method must be called to actually start sampling the item.
///
public bool StartMonitoring(
OperationContext context,
ISampledDataChangeMonitoredItem monitoredItem,
IUserIdentity? savedOwnerIdentity = null)
{
lock (m_lock)
{
if (MeetsGroupCriteria(context, monitoredItem, savedOwnerIdentity))
{
m_itemsToAdd.Add(monitoredItem);
monitoredItem.SetSamplingInterval(m_samplingInterval);
return true;
}
return false;
}
}
///
/// Checks if the monitored item can still be handled by the group.
///
///
/// False if the item has be marked for removal from the group.
///
///
/// The ApplyChanges() method must be called to actually stop sampling the item.
///
public bool ModifyMonitoring(
OperationContext context,
ISampledDataChangeMonitoredItem monitoredItem)
{
lock (m_lock)
{
if (m_items.ContainsKey(monitoredItem.Id))
{
if (MeetsGroupCriteria(context, monitoredItem))
{
monitoredItem.SetSamplingInterval(m_samplingInterval);
return true;
}
m_itemsToRemove.Add(monitoredItem);
}
return false;
}
}
///
/// Stops monitoring the item.
///
///
/// Returns true if the items was marked for removal from the group.
///
public bool StopMonitoring(ISampledDataChangeMonitoredItem monitoredItem)
{
lock (m_lock)
{
if (m_items.ContainsKey(monitoredItem.Id))
{
m_itemsToRemove.Add(monitoredItem);
return true;
}
return false;
}
}
///
/// Updates the group by apply any pending changes.
///
///
/// Returns true if the group has no more items and can be dropped.
///
public bool ApplyChanges()
{
lock (m_lock)
{
// add items.
var itemsToSample = new List();
for (int ii = 0; ii < m_itemsToAdd.Count; ii++)
{
ISampledDataChangeMonitoredItem monitoredItem = m_itemsToAdd[ii];
if (m_items.TryAdd(monitoredItem.Id, monitoredItem) &&
monitoredItem.MonitoringMode != MonitoringMode.Disabled)
{
itemsToSample.Add(monitoredItem);
}
}
m_itemsToAdd.Clear();
// collect first sample.
if (itemsToSample.Count > 0)
{
_ = Task.Run(
async () => await DoSampleAsync(itemsToSample, CancellationToken.None)
.ConfigureAwait(false));
}
// remove items.
for (int ii = 0; ii < m_itemsToRemove.Count; ii++)
{
m_items.Remove(m_itemsToRemove[ii].Id);
}
m_itemsToRemove.Clear();
// start the group if it is not running.
if (m_samplingTask == null && m_items.Count > 0)
{
Startup();
}
// stop the group if it is running.
else if (m_items.Count == 0)
{
Shutdown();
}
// can be shutdown if no items left.
return m_items.Count == 0;
}
}
///
/// Checks if the item meets the group's criteria.
///
private bool MeetsGroupCriteria(
OperationContext context,
ISampledDataChangeMonitoredItem monitoredItem,
IUserIdentity? savedOwnerIdentity = null)
{
// can only sample variables.
if ((monitoredItem.MonitoredItemType & MonitoredItemTypeMask.DataChange) == 0)
{
return false;
}
// can't sample disabled items.
if (monitoredItem.MonitoringMode == MonitoringMode.Disabled)
{
return false;
}
// check sampling interval.
if (AdjustSamplingInterval(monitoredItem.SamplingInterval) != m_samplingInterval)
{
return false;
}
if (m_session == null && context?.SessionId == null)
{
// fallback to compare user Identity if session is not set.
if (!m_effectiveIdentity!.Equals(savedOwnerIdentity))
{
return false;
}
}
else
{
// compare session
if (context?.SessionId != m_session?.Id)
{
return false;
}
}
// check the diagnostics marks.
return m_diagnosticsMask == (context!.DiagnosticsMask & DiagnosticsMasks.OperationAll);
}
///
/// Ensures the requested sampling interval lines up with one of the supported sampling rates.
///
private double AdjustSamplingInterval(double samplingInterval)
{
foreach (SamplingRateGroup samplingRate in m_samplingRates)
{
// groups are ordered by start rate.
if (samplingInterval <= samplingRate.Start)
{
return samplingRate.Start;
}
// check if within range specified by the group.
double maxSamplingRate = samplingRate.Start;
if (samplingRate.Increment > 0)
{
maxSamplingRate += samplingRate.Increment * samplingRate.Count;
}
if (samplingInterval > maxSamplingRate)
{
continue;
}
// find sampling rate within rate group.
if (samplingInterval == maxSamplingRate)
{
return maxSamplingRate;
}
for (double ii = samplingRate.Start;
ii <= maxSamplingRate;
ii += samplingRate.Increment)
{
if (ii >= samplingInterval)
{
return ii;
}
}
}
return samplingInterval;
}
///
/// Periodically checks if the sessions have timed out.
///
private async ValueTask SampleMonitoredItemsAsync(double samplingInterval, CancellationToken cancellationToken = default)
{
try
{
m_logger.LogTrace("Server: {Name} Thread Started.", Thread.CurrentThread.Name);
int sleepCycle = Convert.ToInt32(samplingInterval, CultureInfo.InvariantCulture);
int timeToWait = sleepCycle;
while (m_server.IsRunning)
{
long startTimestamp = m_timeProvider.GetTimestamp();
// wait till next sample.
if (m_shutdownEvent.WaitOne(timeToWait))
{
break;
}
// get current list of items to sample.
var items = new List();
lock (m_lock)
{
uint disabledItemCount = 0;
Dictionary.Enumerator enumerator =
m_items.GetEnumerator();
while (enumerator.MoveNext())
{
ISampledDataChangeMonitoredItem monitoredItem = enumerator.Current
.Value;
if (monitoredItem.MonitoringMode == MonitoringMode.Disabled)
{
disabledItemCount++;
continue;
}
// check whether the item should be sampled.
//if (!monitoredItem.SamplingIntervalExpired())
//{
// continue;
//}
items.Add(monitoredItem);
}
}
// sample the values.
await DoSampleAsync(items, cancellationToken).ConfigureAwait(false);
int delay = (int)m_timeProvider.GetElapsedTime(startTimestamp).TotalMilliseconds;
timeToWait = sleepCycle;
if (delay > sleepCycle)
{
timeToWait = (2 * sleepCycle) - delay;
if (timeToWait < 0)
{
m_logger.LogWarning(
"WARNING: SamplingGroup cannot sample fast enough. TimeToSample={Delay}ms, SamplingInterval={SleepCycle}ms",
delay,
sleepCycle);
timeToWait = sleepCycle;
}
}
}
m_logger.LogTrace("Server: {Name} Thread Exited Normally.", Thread.CurrentThread.Name);
}
catch (Exception e)
{
m_logger.LogError(e, "Server: SampleMonitoredItems Thread Exited Unexpectedly.");
}
}
///
/// Samples the values of the items.
///
private async ValueTask DoSampleAsync(List items, CancellationToken cancellationToken = default)
{
try
{
// read values for all enabled items.
if (items != null && items.Count > 0)
{
var itemsToRead = new List(items.Count);
var values = new List(items.Count);
var errors = new List(items.Count);
// allocate space for results.
for (int ii = 0; ii < items.Count; ii++)
{
ReadValueId readValueId = items[ii].GetReadValueId();
readValueId.Processed = false;
itemsToRead.Add(readValueId);
values.Add(default);
errors.Add(null!);
}
OperationContext context;
if (m_session != null)
{
context = new OperationContext(m_session, m_diagnosticsMask);
}
else
{
// if session of the Sampling group is not set yet, use the first monitored item to create the context.
IMonitoredItem firstItem = items[0];
context = new OperationContext(firstItem);
m_session = firstItem.Session;
}
// read values.
await m_nodeManager.ReadAsync(context, 0, itemsToRead, values, errors, cancellationToken).ConfigureAwait(false);
// update monitored items.
for (int ii = 0; ii < items.Count; ii++)
{
if (values[ii].IsNull)
{
values[ii] = DataValue.FromStatusCode(
StatusCodes.BadInternalError,
m_timeProvider.GetUtcNow().UtcDateTime);
}
items[ii].QueueValue(values[ii], errors[ii]);
}
}
}
catch (Exception e)
{
m_logger.LogError(e, "Server: Unexpected error sampling values.");
}
}
private readonly Lock m_lock = new();
private readonly ILogger m_logger;
private readonly IServerInternal m_server;
private readonly TimeProvider m_timeProvider;
private readonly IAsyncNodeManager m_nodeManager;
private ISession? m_session;
private readonly IUserIdentity? m_effectiveIdentity;
private readonly DiagnosticsMasks m_diagnosticsMask;
private readonly double m_samplingInterval;
private readonly List m_itemsToAdd;
private readonly List m_itemsToRemove;
private readonly Dictionary m_items;
private readonly ManualResetEvent m_shutdownEvent;
private readonly List m_samplingRates;
private Task? m_samplingTask;
}
}