//******************************************************************************************************//  FrameQueue.cs - Gbtc////  Copyright © 2010, 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 Eclipse Public License -v 1.0 (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/eclipse-1.0.php////  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.//  11/18/2010 - Mehulbhai P Thakkar//       Fixed bug in the Pop() method by checking if m_frameQueue has any element in it.//  02/08/2011 - J. Ritchie Carroll//       Added ExamineQueueState method to analyze real-time queue state.//  05/10/2011 - J. Ritchie Carroll//       Updated frame queue locks to use a ReaderWriterSpinLock as an optimization.////****************************************************************************************************** using System;using System.Collections.Concurrent;using System.Collections.Generic;using System.Text;using System.Threading; namespace GSF.TimeSeries{    /// <summary>    /// Represents a real-time queue of <see cref="TrackingFrame"/> instances used by the <see cref="ConcentratorBase"/> class.    /// </summary>    internal class FrameQueue : IDisposable    {        #region [ Members ]         // Delegates        internal delegate IFrame CreateNewFrameFunction(Ticks timestamp);         // Fields        private CreateNewFrameFunction m_createNewFrame;                // IFrame creation function        private LinkedList<TrackingFrame> m_frameList;                  // We keep this list sorted by timestamp so frames are processed in order        private ConcurrentDictionary<long, TrackingFrame> m_frameHash;  // Fast frame lookup dictionary        private SpinLock m_queueLock;                                   // Spinning lock used for synchronizing access to frame list        private long m_publishedTicks;                                  // Timstamp of last published frame        private volatile TrackingFrame m_head;                          // Reference to current top of the frame collection        private volatile TrackingFrame m_last;                          // Reference to last published frame        private int m_framesPerSecond;                                  // Cached frames per second        private double m_ticksPerFrame;                                 // Cached ticks per frame        private long m_timeResolution;                                  // Cached time resolution (max sorting resolution in ticks)        private DownsamplingMethod m_downsamplingMethod;                // Cached downsampling method        private bool m_disposed;                                        // Object disposed flag         #endregion         #region [ Constructors ]         /// <summary>        /// Creates a new <see cref="FrameQueue"/>.        /// </summary>        internal FrameQueue(CreateNewFrameFunction createNewFrame)        {            m_createNewFrame = createNewFrame;            m_frameList = new LinkedList<TrackingFrame>();            m_frameHash = new ConcurrentDictionary<long, TrackingFrame>();            m_queueLock = new SpinLock();            m_downsamplingMethod = DownsamplingMethod.LastReceived;        }         /// <summary>        /// Releases the unmanaged resources before the <see cref="FrameQueue"/> object is reclaimed by <see cref="GC"/>.        /// </summary>        ~FrameQueue()        {            Dispose(false);        }         #endregion         #region [ Properties ]          /// <summary>        /// Gets or sets the number of frames per second to be used by <see cref="FrameQueue"/>.        /// </summary>        /// <remarks>        /// Valid frame rates are greater than 0 frames per second.        /// </remarks>        public int FramesPerSecond        {            get            {                return m_framesPerSecond;            }            set            {                if (value < 1)                    throw new ArgumentOutOfRangeException("value", "Frames per second must be greater than 0");                 m_framesPerSecond = value;                m_ticksPerFrame = Ticks.PerSecond / (double)m_framesPerSecond;            }        }         /// <summary>        /// Gets or sets the maximum time resolution to use when sorting measurements by timestamps into their proper destination frame.        /// </summary>        public long TimeResolution        {            get            {                return m_timeResolution;            }            set            {                m_timeResolution = value;            }        }         /// <summary>        /// Gets or sets the <see cref="GSF.TimeSeries.DownsamplingMethod"/> to be used by the <see cref="FrameQueue"/>.        /// </summary>        public DownsamplingMethod DownsamplingMethod        {            get            {                return m_downsamplingMethod;            }            set            {                m_downsamplingMethod = value;            }        }         /// <summary>        /// Returns the next <see cref="TrackingFrame"/> in the <see cref="FrameQueue"/>, if any.        /// </summary>        /// <remarks>        /// This property is tracked separately from the internal frame collection, as a        /// result this property may be called at any time without a locking penalty.        /// </remarks>        public TrackingFrame Head        {            get            {                // We track the head separately to avoid lock on frame list to safely access first item...                return m_head;            }        }         /// <summary>        /// Gets the last processed <see cref="TrackingFrame"/> in the <see cref="FrameQueue"/>.        /// </summary>        /// <remarks>        /// This property is tracked separately from the internal frame collection, as a        /// result this property may be called at any time without a locking penalty.        /// </remarks>        public TrackingFrame Last        {            get            {                return m_last;            }        }         /// <summary>        /// Returns the total number of frames in the <see cref="FrameQueue"/>.        /// </summary>        public int Count        {            get            {                return m_frameList.Count;            }        }         #endregion         #region [ Methods ]         /// <summary>        /// Releases all the resources used by the <see cref="FrameQueue"/> object.        /// </summary>        public void Dispose()        {            Dispose(true);            GC.SuppressFinalize(this);        }         /// <summary>        /// Releases the unmanaged resources used by the <see cref="FrameQueue"/> 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>        protected virtual void Dispose(bool disposing)        {            if (!m_disposed)            {                try                {                    if (disposing)                    {                        Clear();                        m_frameList = null;                        m_frameHash = null;                         m_createNewFrame = null;                        m_head = null;                        m_last = null;                    }                }                finally                {                    m_disposed = true;  // Prevent duplicate dispose.                }            }        }         /// <summary>        /// Examines and returns the status of the <see cref="FrameQueue"/>.        /// </summary>        /// <param name="expectedMeasurements">Number of expected measurements per frame.</param>        public string ExamineQueueState(int expectedMeasurements)        {            StringBuilder status = new StringBuilder();            bool locked = false;             try            {                m_queueLock.Enter(ref locked);                 status.AppendLine("Concentrator frame queue detail:");                status.AppendLine();                status.AppendLine();                status.AppendFormat(" Ordered frame queue count: {0}", m_frameList.Count);                status.AppendLine();                status.AppendFormat("     Frame hashtable count: {0}", m_frameHash.Count);                status.AppendLine();                 if (m_frameList.Count > 0)                {                    LinkedListNode<TrackingFrame> node = m_frameList.First;                    IFrame frame;                    status.AppendLine();                     for (int i = 0; i < m_frameList.Count; i++)                    {                        if (node.Value != null)                            frame = node.Value.SourceFrame;                        else                            frame = null;                         if (frame == null)                            status.AppendFormat("Frame {0} @ <null frame>", i.ToString().PadLeft(4, '0'));                        else                            status.AppendFormat("Frame {0} @ {1} - {2} measurements, {3} received",                                i.ToString().PadLeft(4, '0'),                                (new DateTime(frame.Timestamp)).ToString("dd-MMM-yyyy HH:mm:ss.fff"),                                frame.Measurements.Count,                                (frame.Measurements.Count / (double)expectedMeasurements).ToString("##0.00%"));                         status.AppendLine();                        node = node.Next;                    }                }                 return status.ToString();            }            finally            {                if (locked)                    m_queueLock.Exit();            }        }         /// <summary>        /// Clears the <see cref="FrameQueue"/>.        /// </summary>        public void Clear()        {            bool locked = false;             try            {                m_queueLock.Enter(ref locked);                 if (m_frameList != null)                    m_frameList.Clear();                 if (m_frameHash != null)                    m_frameHash.Clear();            }            finally            {                if (locked)                    m_queueLock.Exit();            }        }         /// <summary>        /// Removes current <see cref="Head"/> frame from the <see cref="FrameQueue"/> after it has been processed and assigns a new <see cref="Head"/>.        /// </summary>        public void Pop()        {            // We track latest published ticks - don't want to allow slow moving measurements            // to inject themselves after a certain publication timeframe has passed - this            // avoids any possible out-of-sequence frame publication...            m_last = m_head;            m_head = null;            long publishedTicks = m_last.Timestamp;            Thread.VolatileWrite(ref m_publishedTicks, publishedTicks);            bool locked = false;             // Assign next node, if any, as quickly as possible. Still have to wait for queue            // lock - tick-tock, time's-a-wastin' and user function needs a frame to publish.            try            {                m_queueLock.Enter(ref locked);                 if (m_frameList.Count > 0)                {                    LinkedListNode<TrackingFrame> nextNode = m_frameList.First.Next;                     // If next frame is available, go ahead and assign it...                    if (nextNode != null)                        m_head = nextNode.Value;                     // Clean up frame queues                    m_frameList.RemoveFirst();                }            }            finally            {                if (locked)                    m_queueLock.Exit();            }             TrackingFrame frame;            m_frameHash.TryRemove(publishedTicks, out frame);        }         /// <summary>        /// Gets <see cref="TrackingFrame"/> from the queue with the specified timestamp, in ticks.  If no frame exists for        /// the specified timestamp, one will be created.        /// </summary>        /// <param name="ticks">Timestamp, in ticks, for which to get or create <see cref="TrackingFrame"/>.</param>        /// <remarks>        /// Ticks can be any point in time so long time requested is greater than time of last published frame; this queue is        /// used in a real-time scenario with time moving forward.  If a frame is requested for an old timestamp, null will        /// be returned. Note that frame returned will be "best-fit" for given timestamp based on <see cref="FramesPerSecond"/>.        /// </remarks>        /// <returns>An existing or new <see cref="TrackingFrame"/> from the queue for the specified timestamp.</returns>        public TrackingFrame GetFrame(long ticks)        {            // Calculate destination ticks for this frame            TrackingFrame frame = null;            bool locked = false, nodeAdded = false;            long baseTicks, ticksBeyondSecond, frameIndex, destinationTicks, nextDestinationTicks;             // Baseline timestamp to the top of the second            baseTicks = ticks - ticks % Ticks.PerSecond;             // Remove the seconds from ticks            ticksBeyondSecond = ticks - baseTicks;             // Calculate a frame index between 0 and m_framesPerSecond-1, corresponding to ticks            // rounded down to the nearest frame            frameIndex = (long)(ticksBeyondSecond / m_ticksPerFrame);             // Calculate the timestamp of the nearest frame rounded up            nextDestinationTicks = (frameIndex + 1) * Ticks.PerSecond / m_framesPerSecond;             // Determine whether the desired frame is the nearest            // frame rounded down or the nearest frame rounded up            if (m_timeResolution <= 1)            {                if (nextDestinationTicks <= ticksBeyondSecond)                    destinationTicks = nextDestinationTicks;                else                    destinationTicks = frameIndex * Ticks.PerSecond / m_framesPerSecond;            }            else            {                // If, after translating nextDestinationTicks to the time resolution, it is less than                // or equal to ticks, nextDestinationTicks corresponds to the desired frame                if ((nextDestinationTicks / m_timeResolution) * m_timeResolution <= ticksBeyondSecond)                    destinationTicks = nextDestinationTicks;                else                    destinationTicks = frameIndex * Ticks.PerSecond / m_framesPerSecond;            }             // Recover the seconds that were removed            destinationTicks += baseTicks;             // Make sure ticks are newer than latest published ticks...            if (destinationTicks > Thread.VolatileRead(ref m_publishedTicks))            {                // See if requested frame is already available (can do this outside lock with concurrent dictionary)                if (m_frameHash.TryGetValue(destinationTicks, out frame))                    return frame;                 // Didn't find frame for this timestamp so we need to add a new one to the queue                try                {                    m_queueLock.Enter(ref locked);                     // Another thread may have gotten to this task already, so check for this contingency...                    if (m_frameHash.TryGetValue(destinationTicks, out frame))                        return frame;                     // Create a new frame for this timestamp                    frame = new TrackingFrame(m_createNewFrame(destinationTicks), m_downsamplingMethod);                     if (m_frameList.Count > 0)                    {                        // Insert frame into proper sorted position...                        LinkedListNode<TrackingFrame> node = m_frameList.Last;                         do                        {                            if (destinationTicks > node.Value.Timestamp)                            {                                m_frameList.AddAfter(node, frame);                                nodeAdded = true;                                break;                            }                             node = node.Previous;                        }                        while (node != null);                    }                     if (!nodeAdded)                    {                        m_frameList.AddFirst(frame);                        m_head = frame;                    }                     // Since we'll be requesting this frame over and over, we'll use                    // a hash table for quick frame lookups by timestamp                    m_frameHash[destinationTicks] = frame;                }                finally                {                    if (locked)                        m_queueLock.Exit();                }            }             return frame;        }         #region [ GetFrame Testing Algorithm ]         // Copy this code into a console application and reference GSF.Core.dll to test.         //using System;        //using System.Collections.Generic;        //using System.Linq;        //using System.Text;        //using GSF;         //namespace FrameTimestampTest        //{        //    public class Program        //    {        //        private static long s_timeResolution;        //        private static long s_framesPerSecond;        //        private static double s_ticksPerFrame;        //        private static double s_millisecondsPerFrame;         //        public static void Main(string[] args)        //        {        //            s_framesPerSecond = 30;        //            s_ticksPerFrame = Ticks.PerSecond / (double)s_framesPerSecond;        //            s_timeResolution = Ticks.PerMillisecond;        //            s_millisecondsPerFrame = 1000.0 / s_framesPerSecond;         //            // Test BPA PDCstream style milliseconds        //            DateTime currentTime = DateTime.Now;        //            int[] bpaMilliseconds = new int[] { 000, 033, 066, 100, 133, 166, 200, 233, 266, 300, 333, 366, 400, 433, 466, 500, 533, 566, 599, 633, 666, 699, 733, 766, 800, 833, 866, 900, 933, 966 };         //            s_timeResolution = Ticks.PerMillisecond * 33;        //            Console.WriteLine("BPA millisecond sorting test, TimeResolution = {0}", s_timeResolution);         //            for (int i = 0; i < bpaMilliseconds.Length; i++)        //            {        //                long longTicks = (new DateTime(currentTime.Year, currentTime.Month, currentTime.Day, currentTime.Hour, currentTime.Minute, currentTime.Second, bpaMilliseconds[i])).Ticks;        //                long destination = GetFrame(longTicks);        //                Console.WriteLine(string.Format("{0:000} ms : {1:000} ms", bpaMilliseconds[i], (new DateTime(destination)).Millisecond));        //            }         //            s_timeResolution = Ticks.PerMillisecond;

Lines 1–500 of 595. Load more