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