//******************************************************************************************************//  ConcentratorBase.cs - Gbtc////  Copyright © 2012, Grid Protection Alliance.  All Rights Reserved.////  Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See//  the NOTICE file distributed with this work for additional information regarding copyright ownership.//  The GPA licenses this file to you under the MIT License (MIT), the "License"; you may//  not use this file except in compliance with the License. You may obtain a copy of the License at:////      http://www.opensource.org/licenses/MIT////  Unless agreed to in writing, the subject software distributed under the License is distributed on an//  "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the//  License for the specific language governing permissions and limitations.////  Code Modification History://  ----------------------------------------------------------------------------------------------------//  09/02/2010 - J. Ritchie Carroll//       Generated original version of source code.//  02/08/2011 - J. Ritchie Carroll//       Added QueueState property to expose real-time queue state analysis.//  04/14/2011 - J. Ritchie Carroll//       Added ProcessByReceivedTimestamp property to sort and publish measurements by received time.//  05/06/2011 - J. Ritchie Carroll / Jian (Ryan) Zuo//       Updated to reduce lock contention and added volatile reads on stats for better accuracy.//  12/20/2012 - Starlynn Danyelle Gilliam//       Modified Header.////****************************************************************************************************** using System;using System.Collections.Generic;using System.Linq;using System.Text;using System.Threading;using GSF.Diagnostics;using GSF.Threading;using GSF.Units; // ReSharper disable PossibleMultipleEnumerationnamespace GSF.TimeSeries{    #region [ Enumerations ]     /// <summary>    /// Down-sampling method enumeration.    /// </summary>    public enum DownsamplingMethod    {        /// <summary>        /// Down-samples to the last measurement received.        /// </summary>        /// <remarks>        /// Use this option if no down-sampling is needed or the selected value is not critical. This is the fastest option if the incoming and outgoing frame rates match.        /// </remarks>        LastReceived,        /// <summary>        /// Down-samples to the measurement closest to frame time.        /// </summary>        /// <remarks>        /// This is the typical operation used when performing simple down-sampling. This is the fastest option if the incoming frame rate is faster than the outgoing frame rate.        /// </remarks>        Closest,        /// <summary>        /// Down-samples by applying a user-defined value filter over all received measurements to anti-alias the results.        /// </summary>        /// <remarks>        /// This option will produce the best result but has a processing penalty.        /// </remarks>        Filtered,        /// <summary>        /// Down-samples to the measurement that has the best quality closest to frame time.        /// </summary>        /// <remarks>        /// This option chooses the measurement closest to the frame time with the best quality.        /// </remarks>        BestQuality    }     #endregion     /// <summary>    /// Measurement concentrator base class.    /// </summary>    /// <remarks>    /// <para>    /// This class synchronizes (i.e., sorts by timestamp) real-time measurements.    /// </para>    /// <para>    /// Note that your lag time should be defined as it relates to the rate at which data is coming    /// into the concentrator. Make sure you allow enough time for transmission of data over the network    /// allowing any needed time for possible network congestion.  Lead time should be defined as your    /// confidence in the accuracy of your local clock (e.g., if you set lead time to 2, this means you    /// trust that your local clock is within plus or minus 2 seconds of real-time.)    /// </para>    /// <para>    /// This concentrator is designed to sort measurements being transmitted in real-time for data being    /// sent at rates of at least 1 sample per second. Slower rates (e.g., once every few seconds) are not    /// supported since sorting data at these speeds would be trivial. There is no defined maximum number    /// of supported samples per second - but keep in mind that CPU utilization will increase as the    /// measurement volume and frame rate increase.    /// </para>    /// </remarks>    public abstract class ConcentratorBase : IDisposable    {        #region [ Members ]         // Nested Types         /// <summary>        /// Frame rate timer.        /// </summary>        /// <remarks>        /// One static instance of this internal class is created per encountered frame rate / processing interval.        /// </remarks>        private sealed class FrameRateTimer : IDisposable        {            #region [ Members ]             // Fields            private PrecisionTimer m_timer;            private readonly int m_framesPerSecond;            private readonly int m_processingInterval;            private readonly int[] m_framePeriods;            private int m_frameIndex;            private int m_lastFramePeriod;            private int m_referenceCount;            private bool m_disposed;             #endregion             #region [ Constructors ]             /// <summary>            /// Create a new <see cref="FrameRateTimer"/> class.            /// </summary>            /// <param name="framesPerSecond">Desired frame rate for <see cref="PrecisionTimer"/>.</param>            /// <param name="processingInterval">Desired processing interval, if applicable.</param>            /// <remarks>            /// When the <paramref name="processingInterval"/> is set to -1, the frame rate timer interval will be calculated as a distribution            /// of whole milliseconds over the specified number of <paramref name="framesPerSecond"/>. Otherwise the specified            /// <paramref name="processingInterval"/> will be used as the timer interval.            /// </remarks>            public FrameRateTimer(int framesPerSecond, int processingInterval)            {                if (processingInterval == 0)                    throw new InvalidOperationException("A frame rate timer should not be created when using a processing interval of zero, i.e., processing data as fast as possible.");                 m_framesPerSecond = framesPerSecond;                m_processingInterval = processingInterval;                 // Create a new precision timer for this timer state                m_timer = new PrecisionTimer();                m_timer.AutoReset = true;                 if (processingInterval > 0)                {                    // Establish fixed timer period                    m_timer.Period = processingInterval;                }                else                {                    // Attach handler for timer period assignments                    m_timer.Tick += SetTimerPeriod;                     // Calculate distributed wait time periods over specified number of frames per second                    m_framePeriods = new int[framesPerSecond];                     for (int frameIndex = 0; frameIndex < framesPerSecond; frameIndex++)                    {                        m_framePeriods[frameIndex] = CalcWaitTimeForFrameIndex(frameIndex);                    }                     // Establish initial timer period                    m_lastFramePeriod = m_framePeriods[0];                    m_timer.Period = m_lastFramePeriod;                }                 // Start timer                m_timer.Start();            }             /// <summary>            /// Releases the unmanaged resources before the <see cref="FrameRateTimer"/> object is reclaimed by <see cref="GC"/>.            /// </summary>            ~FrameRateTimer()            {                Dispose(false);            }             #endregion             #region [ Properties ]             /// <summary>            /// Gets frames per second for this <see cref="FrameRateTimer"/>.            /// </summary>            public int FramesPerSecond => m_framesPerSecond;             /// <summary>            /// Gets the processing interval defined for the <see cref="FrameRateTimer"/>.            /// </summary>            public int ProcessingInterval => m_processingInterval;             /// <summary>            /// Gets reference count for this <see cref="FrameRateTimer"/>.            /// </summary>            public int ReferenceCount => m_referenceCount;             #endregion             #region [ Methods ]             /// <summary>            /// Releases all the resources used by the <see cref="FrameRateTimer"/> object.            /// </summary>            public void Dispose()            {                Dispose(true);                GC.SuppressFinalize(this);            }             /// <summary>            /// Releases the unmanaged resources used by the <see cref="FrameRateTimer"/> object and optionally releases the managed resources.            /// </summary>            /// <param name="disposing">true to release both managed and unmanaged resources; false to release only unmanaged resources.</param>            private void Dispose(bool disposing)            {                if (!m_disposed)                {                    try                    {                        if (disposing)                        {                            if ((object)m_timer != null)                            {                                if (m_processingInterval == -1)                                    m_timer.Tick -= SetTimerPeriod;                                 m_timer.Dispose();                            }                            m_timer = null;                        }                    }                    finally                    {                        m_disposed = true;  // Prevent duplicate dispose.                    }                }            }             /// <summary>            /// Adds a reference to this <see cref="FrameRateTimer"/>.            /// </summary>            /// <param name="tickFunction">Tick function to add to event list.</param>            public void AddReference(EventHandler tickFunction)            {                m_timer.Tick += tickFunction;                m_referenceCount++;            }             /// <summary>            /// Removes a reference to this <see cref="FrameRateTimer"/>.            /// </summary>            /// <param name="tickFunction">Tick function to remove from event list.</param>            public void RemoveReference(EventHandler tickFunction)            {                m_timer.Tick -= tickFunction;                m_referenceCount--;            }             // Handler to assign next timer period            private void SetTimerPeriod(object sender, EventArgs e)            {                int period;                 // First things first, prepare timer period for next call...                m_frameIndex++;                 if (m_frameIndex >= m_framesPerSecond)                    m_frameIndex = 0;                 // Get the frame period for this frame index                period = m_framePeriods[m_frameIndex];                 // We only update timer period if it has changed since last call. Note that this is necessary since                // timer periods are defined as integers but actual period is typically uneven (e.g., 33.333 ms)                if (m_lastFramePeriod != period)                    m_timer.Period = period;                 m_lastFramePeriod = period;            }             // Wait times are not necessarily perfectly even (e.g., at 30 samples per second wait time per frame is 33.333... milliseconds)            // so we use this function to evenly distribute integer based millisecond wait times across a second.            private int CalcWaitTimeForFrameIndex(int frameIndex)            {                // Jian Zuo...                int millisecondsWaitTime;                int frameRate;                int deficit;                 frameRate = (int)Math.Round(1000.0D / m_framesPerSecond);                deficit = 1000 - frameRate * m_framesPerSecond;                 if (deficit == 0)                {                    millisecondsWaitTime = frameRate;                }                else                {                    if (frameIndex == 0)                    {                        millisecondsWaitTime = frameRate;                    }                    else if (frameIndex == m_framesPerSecond - 1)                    {                        millisecondsWaitTime = frameRate + (deficit > 0 ? 1 : -1);                    }                    else                    {                        double interval = m_framesPerSecond / Math.Abs((double)deficit);                        double pre_dis = mod_dis(frameIndex - 1, interval);                        double cur_dis = mod_dis(frameIndex, interval);                        double next_dis = mod_dis(frameIndex + 1, interval);                         millisecondsWaitTime = frameRate + (cur_dis <= pre_dis && cur_dis < next_dis ? (deficit > 0 ? 1 : -1) : 0);                    }                }                 return millisecondsWaitTime;            }             private double mod_dis(int framesIndex, double interval)            {                double dis2 = (framesIndex + 1) % interval;                double dis1 = interval - dis2;                return dis1 < dis2 ? dis1 : dis2;            }             #endregion        }         // Events         /// <summary>        /// This event is raised every 5 seconds allowing consumer to track current number of unpublished seconds of data in the queue.        /// </summary>        /// <remarks>        /// <see cref="EventArgs{T}.Argument"/> is the total number of unpublished seconds of data.        /// </remarks>        public event EventHandler<EventArgs<int>> UnpublishedSamples;         /// <summary>        /// This event is raised if there is an exception encountered while attempting to process a frame in the sample queue.        /// </summary>        /// <remarks>        /// <para>        /// <see cref="EventArgs{T}.Argument"/> is the <see cref="Exception"/> encountered while parsing the data stream.        /// </para>        /// <para>        /// Processing will not stop for any exceptions thrown by the user function, but any captured exceptions will be exposed through this event.        /// </para>        /// </remarks>        public event EventHandler<EventArgs<Exception>> ProcessException;         /// <summary>        /// This event is raised if there are any measurements being discarded during the sorting process.        /// </summary>        /// <remarks>        /// <see cref="EventArgs{T}.Argument"/> is the enumeration of <see cref="IMeasurement"/> values that are being discarded during the sorting process.        /// </remarks>        public event EventHandler<EventArgs<IEnumerable<IMeasurement>>> DiscardingMeasurements;         /// <summary>        /// This event is raised when <see cref="ConcentratorBase"/> is disposed.        /// </summary>        public event EventHandler Disposed;         /// <summary>        /// Raised, for the benefit of dependent classes, when lag time is updated.        /// </summary>         internal event Action<double> LagTimeUpdated;         /// <summary>        /// Raised, for the benefit of dependent classes, when lead time is updated.        /// </summary>         internal event Action<double> LeadTimeUpdated;         // Fields        private FrameQueue m_frameQueue;                    // Queue of frames to be published        private Thread m_publicationThread;                 // Thread that handles frame publication        private AutoResetEvent m_publicationWaitHandle;     // Inter-frame publication wait handle        private bool m_usePrecisionTimer;                   // Flag that enables use of precision timer (over just simple thread sleep)        private bool m_attachedToFrameRateTimer;            // Flag that tracks if instance is attached to a frame rate timer        private SharedTimer m_monitorTimer;                 // Sample monitor - tracks total number of unpublished frames        private int m_framesPerSecond;                      // Frames per second        private double m_ticksPerFrame;                     // Ticks per frame        private double m_lagTime;                           // Allowed past time deviation tolerance, in seconds        private double m_leadTime;                          // Allowed future time deviation tolerance, in seconds        private long m_timeResolution;                      // Maximum sorting resolution in ticks        private bool m_roundToNearestTimestamp;             // Determines whether to round to nearest timestamp        private int m_processingInterval;                   // Defines a specific processing interval for data, if desired        private DownsamplingMethod m_downsamplingMethod;    // Down-sampling method to use if input is at a higher-resolution than output        private double m_timeOffset;                        // Half the distance of the time resolution used for index calculation        private int m_maximumPublicationTimeout;            // Maximum publication wait timeout        private Ticks m_lagTicks;                           // Current lag time calculated in ticks        private volatile bool m_enabled;                    // Enabled state of concentrator        private long m_startTime;                           // Start time of concentrator        private long m_stopTime;                            // Stop time of concentrator        private long m_realTimeTicks;                       // Timestamp of real-time or the most recently received measurement        private bool m_ignoreBadTimestamps;                 // Determines whether or not to ignore bad timestamps when sorting measurements        private bool m_allowSortsByArrival;                 // Determines whether or not to sort incoming measurements with a bad timestamp by arrival        private bool m_useLocalClockAsRealTime;             // Determines whether or not to use local system clock as "real-time"        private bool m_allowPreemptivePublishing;           // Determines whether or not to preemptively publish frame if expected measurements arrive        private bool m_performTimestampReasonabilityCheck;  // Determines whether or not to execute timestamp reasonability checks (i.e., lead time validation)        private bool m_processByReceivedTimestamp;          // Determines whether or not to sort and publish measurements by their ReceivedTimestamp        private int m_expectedMeasurements;                 // Expected number of measurements to be sorted into a frame        private long m_receivedMeasurements;                // Total number of measurements ever received for sorting        private long m_processedMeasurements;               // Total number of measurements ever successfully sorted        private long m_discardedMeasurements;               // Total number of discarded measurements        private long m_measurementsSortedByArrival;         // Total number of measurements that were sorted by arrival        private long m_publishedMeasurements;               // Total number of published measurements        private long m_downsampledMeasurements;             // Total number of down-sampled measurements        private long m_missedSortsByTimeout;                // Total number of unsorted measurements due to timeout waiting for lock        private long m_waitHandleExpirations;               // Total number of wait handle expirations encountered due to delayed precision timer releases        private long m_framesAheadOfSchedule;               // Total number of frames published ahead of schedule        private long m_publishedFrames;                     // Total number of published frames        private long m_totalPublishTime;                    // Total cumulative frame user function publication time (in ticks) - used to calculate average        private bool m_trackLatestMeasurements;             // Determines whether or not to track latest measurements        private ImmediateMeasurements m_latestMeasurements; // Absolute latest received measurement values        private IMeasurement m_lastDiscardedMeasurement;    // Last measurement that was discarded by the concentrator        private long m_lastDiscardedMeasurementLatency;     // Latency of last measurement that was discarded by the concentrator        private bool m_disposed;                            // Disposed flag detects redundant calls to dispose method         #endregion         #region [ Constructors ]         /// <summary>        /// Creates a new <see cref="ConcentratorBase"/>.        /// </summary>        /// <remarks>        /// Concentration will not begin until consumer "Starts" concentrator (i.e., calling <see cref="ConcentratorBase.Start"/> method or setting        /// <c><see cref="ConcentratorBase.Enabled"/> = true</c>).        /// </remarks>        protected ConcentratorBase()        {            Log = Logger.CreatePublisher(GetType(), MessageClass.Framework);            m_usePrecisionTimer = true;            m_allowSortsByArrival = true;            m_allowPreemptivePublishing = true;            m_performTimestampReasonabilityCheck = true;            m_processingInterval = -1;            m_downsamplingMethod = DownsamplingMethod.LastReceived;            m_latestMeasurements = new ImmediateMeasurements(this);            m_maximumPublicationTimeout = Timeout.Infinite;             // Create a new queue for managing real-time frames            m_frameQueue = new FrameQueue(this.CreateNewFrame);             // Set minimum timer resolution to one millisecond to improve timer accuracy            PrecisionTimer.SetMinimumTimerResolution(1);             // Create publication wait handle            m_publicationWaitHandle = new AutoResetEvent(false);             // Create publication thread            m_publicationThread = new Thread(PublishFrames);            m_publicationThread.Start();             // This timer monitors the total number of unpublished samples every 5 seconds. This is a useful statistic            // to monitor: if total number of unpublished samples exceed lag time, measurement concentration could            // be falling behind.            m_monitorTimer = Common.TimerScheduler.CreateTimer(5000);            m_monitorTimer.AutoReset = true;            m_monitorTimer.Elapsed += MonitorUnpublishedSamples;        }         /// <summary>        /// Creates a new <see cref="ConcentratorBase"/> from specified parameters.        /// </summary>        /// <param name="framesPerSecond">Number of frames to publish per second.</param>        /// <param name="lagTime">Past time deviation tolerance, in seconds - this becomes the amount of time to wait before publishing begins.</param>        /// <param name="leadTime">Future time deviation tolerance, in seconds - this becomes the tolerated +/- accuracy of the local clock to real-time.</param>        /// <remarks>        /// <para>        /// <paramref name="framesPerSecond"/> must be greater then 0.        /// </para>        /// <para>        /// <paramref name="lagTime"/> must be greater than zero, but can be specified in sub-second intervals (e.g., set to .25 for a quarter-second lag time).        /// Note that this defines time sensitivity to past timestamps.        /// </para>        /// <para>        /// <paramref name="leadTime"/> must be greater than zero, but can be specified in sub-second intervals (e.g., set to .5 for a half-second lead time).        /// Note that this defines time sensitivity to future timestamps.        /// </para>        /// <para>        /// Concentration will not begin until consumer "Starts" concentrator (i.e., calling <see cref="ConcentratorBase.Start"/> method or setting

Lines 1–500 of 2205. Load more