Migrated changes from VS2010: 1) GSF.Core: Updated process queue to be based on IProducerConsumerCollection<T> model using a lock-free concurrent queue for common default operations. 2) GSF.Core: Optimized implementation of MultiSourceFrameImageParserBase using concurrent dictionaries. 3) GSF.Core: Added an AddRange method to ProcessQueue base implementation. 4) GSF.TimeSeries: Updated OutputAdapterBase to better take advantage of new ProcessQueue functionality. 5) GSF.Historian: Updated openHistorian 1.0 code to accomodate updates to ProcessQueue.

StephenCWills committed 14 years ago

Commit ae78d0f1 parent cafd09910 files · +2658 −1805Browse files

Showing the changes to Source/Libraries/GSF.Core/Parsing/MultiSourceFrameImageParserBase.cs (1 of 10 files). Show all

Unified · Split

Source/Libraries/GSF.Core/Parsing/MultiSourceFrameImageParserBase.cs+23−38

 //       Edited Comments.
 //  09/14/2009 - Stephen C. Wills
 //       Added new header and license agreement.
+//  11/01/2012 - J. Ritchie Carroll
+//       Updated to use concurrent dictionaries and new recreate parsed output list.
 //
 //*******************************************************************************************************
 using GSF.Collections;
 using System;
+using System.Collections.Concurrent;
 using System.Collections.Generic;
 using System.ComponentModel;
 using System.Text;
         // Fields
         private ProcessQueue<IdentifiableItem<TSourceIdentifier, byte[]>> m_bufferQueue;
-        private Dictionary<TSourceIdentifier, bool> m_sourceInitialized;
+        private ConcurrentDictionary<TSourceIdentifier, bool> m_sourceInitialized;
-        private Dictionary<TSourceIdentifier, byte[]> m_unparsedBuffers;
+        private ConcurrentDictionary<TSourceIdentifier, byte[]> m_unparsedBuffers;
-        private List<TOutputType> m_parsedOutputs;
+        private readonly List<TOutputType> m_parsedOutputs;
         private TSourceIdentifier m_sourceID;
         private bool m_disposed;
             m_bufferQueue.ProcessException += ProcessExceptionHandler;
             if (ProtocolUsesSyncBytes)
-                m_sourceInitialized = new Dictionary<TSourceIdentifier, bool>();
+                m_sourceInitialized = new ConcurrentDictionary<TSourceIdentifier, bool>();
-            m_unparsedBuffers = new Dictionary<TSourceIdentifier, byte[]>();
+            m_unparsedBuffers = new ConcurrentDictionary<TSourceIdentifier, byte[]>();
+            m_parsedOutputs = new List<TOutputType>();
             // We attach to base class events so we can cumulate outputs per data source and handle data errors
-            base.DataParsed += CumulateParsedOutput;
+            base.DataParsed += DataParsed_CumulateParsedOutput;
-            base.DataDiscarded += ResetDataSourceInitialization;
+            base.DataDiscarded += DataDiscarded_ResetDataSourceInitialization;
         }
         #endregion
                         }
                         m_bufferQueue = null;
-                        if ((object)m_sourceInitialized != null)
-                        {
-                            m_sourceInitialized.Clear();
-                        }
                         m_sourceInitialized = null;
-                        if ((object)m_unparsedBuffers != null)
-                        {
-                            m_unparsedBuffers.Clear();
-                        }
                         m_unparsedBuffers = null;
                         // Detach from base class events
-                        base.DataParsed -= CumulateParsedOutput;
+                        base.DataParsed -= DataParsed_CumulateParsedOutput;
-                        base.DataDiscarded -= ResetDataSourceInitialization;
+                        base.DataDiscarded -= DataDiscarded_ResetDataSourceInitialization;
                     }
                 }
                 finally
         /// </remarks>
         public virtual void PurgeBuffer(TSourceIdentifier source)
         {
-            lock (m_unparsedBuffers)
+            byte[] buffer;
-            {
+            m_unparsedBuffers.TryRemove(source, out buffer);
-                if (m_unparsedBuffers.ContainsKey(source))
-                    m_unparsedBuffers.Remove(source);
-            }
         }
         /// <summary>
                 buffer = item.Item;
                 // Check to see if this data source has been initialized
-                if ((object)m_sourceInitialized != null && !m_sourceInitialized.TryGetValue(m_sourceID, out StreamInitialized))
+                if ((object)m_sourceInitialized != null)
-                    m_sourceInitialized.Add(m_sourceID, true);
+                    StreamInitialized = m_sourceInitialized.GetOrAdd(m_sourceID, true);
                 // Restore any unparsed buffers for this data source, if any
-                lock (m_unparsedBuffers)
+                UnparsedBuffer = m_unparsedBuffers.GetOrAdd(m_sourceID, (byte[])null);
-                {
-                    if (!m_unparsedBuffers.TryGetValue(m_sourceID, out UnparsedBuffer))
-                        m_unparsedBuffers.Add(m_sourceID, null);
-                }
-                // Create new collection for parsed output
+                // Clear collection to hold new parsed output
-                m_parsedOutputs = new List<TOutputType>();
+                m_parsedOutputs.Clear();
                 // Start parsing sequence for this buffer - this will cumulate new parsed outputs
                 base.Write(buffer, 0, buffer.Length);
                 // Track last unparsed buffer for this data source
-                lock (m_unparsedBuffers)
+                m_unparsedBuffers[m_sourceID] = UnparsedBuffer;
-                {
-                    m_unparsedBuffers[m_sourceID] = UnparsedBuffer;
-                }
                 // Expose any parsed data
                 if (m_parsedOutputs.Count > 0)
         }
         // Cumulate output data for current data source
-        private void CumulateParsedOutput(object sender, EventArgs<TOutputType> data)
+        private void DataParsed_CumulateParsedOutput(object sender, EventArgs<TOutputType> data)
         {
             m_parsedOutputs.Add(data.Argument);
         }
         // If an error occurs during parsing from a data source, we reset its initialization state
-        private void ResetDataSourceInitialization(object sender, EventArgs<byte[]> data)
+        private void DataDiscarded_ResetDataSourceInitialization(object sender, EventArgs<byte[]> data)
         {
             if ((object)m_sourceInitialized != null)
                 m_sourceInitialized[m_sourceID] = false;

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code