GSF.TimeSeries: Added a new routing method designed to process measurements in batch and reduce overall locking required.

StevenChisholm committed 12 years ago

Commit ef1de4e1 parent aeab67f3 files · +218 −3Browse files

Unified · Split

3 files

Source/Libraries/GSF.TimeSeries/Adapters/RoutingTableInputBuffer.cs +208 −0
Source/Libraries/GSF.TimeSeries/Adapters/RoutingTables.cs +9 −3
Source/Libraries/GSF.TimeSeries/GSF.TimeSeries.csproj +1 −0

Source/Libraries/GSF.TimeSeries/Adapters/RoutingTableInputBuffer.cs+208−0

@@ -0,0 +1,208 @@
	1	//******************************************************************************************************
	2	// RoutingTableInputBuffer.cs - Gbtc
	3	//
	4	// Copyright © 2014, Grid Protection Alliance. All Rights Reserved.
	5	//
	6	// Licensed to the Grid Protection Alliance (GPA) under one or more contributor license agreements. See
	7	// the NOTICE file distributed with this work for additional information regarding copyright ownership.
	8	// The GPA licenses this file to you under the Eclipse Public License -v 1.0 (the "License"); you may
	9	// not use this file except in compliance with the License. You may obtain a copy of the License at:
	10	//
	11	// http://www.opensource.org/licenses/eclipse-1.0.php
	12	//
	13	// Unless agreed to in writing, the subject software distributed under the License is distributed on an
	14	// "AS-IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied. Refer to the
	15	// License for the specific language governing permissions and limitations.
	16	//
	17	// Code Modification History:
	18	// ----------------------------------------------------------------------------------------------------
	19	// 04/23/2014 - Steven E. Chisholm
	20	// Generated original version of source code.
	21	//
	22	//******************************************************************************************************
	23
	24	using System;
	25	using System.Collections.Generic;
	26	using System.Linq;
	27	using System.Threading;
	28	using GSF.Collections;
	29	using GSF.Threading;
	30
	31	namespace GSF.TimeSeries.Adapters
	32	{
	33	public partial class RoutingTables
	34	{
	35	private class RoutingTableInputBuffer : IDisposable
	36	{
	37	class Buffer
	38	{
	39	public List<List<IMeasurement>> MeasurementList = new List<List<IMeasurement>>();
	40	public List<IEnumerable<IMeasurement>> Measurement = new List<IEnumerable<IMeasurement>>();
	41
	42	public void Enqueue(IEnumerable<IMeasurement> measurements)
	43	{
	44	List<IMeasurement> list = measurements as List<IMeasurement>;
	45	if (list != null)
	46	{
	47	MeasurementList.Add(list);
	48	}
	49	else
	50	{
	51	Measurement.Add(measurements);
	52	}
	53	}
	54
	55	public void Clear()
	56	{
	57	MeasurementList.Clear();
	58	Measurement.Clear();
	59	}
	60	}
	61
	62	Buffer m_activeBuffer;
	63	Buffer m_processingBuffer;
	64
	65	private Dictionary<Guid, List<Producer>> m_localSignalLookup;
	66	private Dictionary<Consumer, Producer> m_localDestinationLookup;
	67	private RoutingTables m_routingTables;
	68	private object m_localCacheLock;
	69	private int m_version;
	70	private ScheduledTask m_task;
	71
	72	public RoutingTableInputBuffer(RoutingTables routingTables)
	73	{
	74	m_activeBuffer = new Buffer();
	75	m_processingBuffer = new Buffer();
	76	m_localCacheLock = new object();
	77	m_localSignalLookup = new Dictionary<Guid, List<Producer>>();
	78	m_localDestinationLookup = new Dictionary<Consumer, Producer>();
	79	m_routingTables = routingTables;
	80	m_task = new ScheduledTask(ThreadingMode.DedicatedBackground, ThreadPriority.Normal);
	81	m_task.Running += m_task_Running;
	82	}
	83
	84	void m_task_Running(object sender, EventArgs<ScheduledTaskRunningReason> e)
	85	{
	86	if (e.Argument == ScheduledTaskRunningReason.Disposing)
	87	return;
	88
	89	lock (m_localCacheLock)
	90	{
	91	Buffer temp = m_activeBuffer;
	92	m_activeBuffer = m_processingBuffer;
	93	m_processingBuffer = temp;
	94	m_activeBuffer.Clear();
	95	}
	96
	97	GlobalCache globalCache;
	98	List<Producer> producers;
	99	List<Consumer> consumers;
	100
	101	// Get the global cache from the routing tables
	102	globalCache = Interlocked.CompareExchange(ref m_routingTables.m_globalCache, null, null);
	103
	104	// Return if routes are still being calculated
	105	if ((object)globalCache == null)
	106	return;
	107
	108	// Check the version of the local cache against that of the global cache.
	109	// We need to clear the local cache if the versions don't match because
	110	// that means routes have changed
	111	if (m_version != globalCache.Version)
	112	{
	113	// Dump the signal lookup
	114	m_localSignalLookup.Clear();
	115
	116	// Best if we hang onto producers for adapters that still have routes
	117	foreach (Consumer consumer in m_localDestinationLookup.Keys.Where(consumer => !globalCache.GlobalDestinationLookup.ContainsKey(consumer.Adapter)).ToList())
	118	{
	119	m_localDestinationLookup[consumer].QueueProducer.Dispose();
	120	m_localDestinationLookup.Remove(consumer);
	121	}
	122
	123	// Update the local cache version
	124	m_version = globalCache.Version;
	125	}
	126
	127	foreach (var group in m_processingBuffer.Measurement)
	128	{
	129	//Exact same code as below
	130	foreach (IMeasurement measurement in group)
	131	{
	132	// Attempt to look up the signal in the local cache
	133	if (!m_localSignalLookup.TryGetValue(measurement.ID, out producers))
	134	{
	135	// Not in the local cache - check the global cache and fall back on broadcast consumers
	136	if (!globalCache.GlobalSignalLookup.TryGetValue(measurement.ID, out consumers))
	137	consumers = globalCache.BroadcastConsumers;
	138
	139	// Get a producer for each of the consumers
	140	producers = consumers
	141	.Select(consumer => m_localDestinationLookup.GetOrAdd(consumer, c => new Producer(c.Manager)))
	142	.ToList();
	143
	144	// Add this signal to the local cache
	145	m_localSignalLookup.Add(measurement.ID, producers);
	146	}
	147
	148	// Add this measurement to the producers' list
	149	foreach (Producer producer in producers)
	150	producer.Measurements.Add(measurement);
	151	}
	152	}
	153
	154	foreach (var group in m_processingBuffer.MeasurementList)
	155	{
	156	//Exact same code as above
	157	foreach (IMeasurement measurement in group)
	158	{
	159	// Attempt to look up the signal in the local cache
	160	if (!m_localSignalLookup.TryGetValue(measurement.ID, out producers))
	161	{
	162	// Not in the local cache - check the global cache and fall back on broadcast consumers
	163	if (!globalCache.GlobalSignalLookup.TryGetValue(measurement.ID, out consumers))
	164	consumers = globalCache.BroadcastConsumers;
	165
	166	// Get a producer for each of the consumers
	167	producers = consumers
	168	.Select(consumer => m_localDestinationLookup.GetOrAdd(consumer, c => new Producer(c.Manager)))
	169	.ToList();
	170
	171	// Add this signal to the local cache
	172	m_localSignalLookup.Add(measurement.ID, producers);
	173	}
	174
	175	// Add this measurement to the producers' list
	176	foreach (Producer producer in producers)
	177	producer.Measurements.Add(measurement);
	178	}
	179	}
	180
	181	// Produce measurements to consumers in the local destination
	182	// cache which have measurements to be received
	183	foreach (Producer producer in m_localDestinationLookup.Values)
	184	{
	185	if (producer.Measurements.Count > 0)
	186	{
	187	producer.QueueProducer.Produce(producer.Measurements);
	188	producer.Measurements.Clear();
	189	}
	190	}
	191	}
	192
	193	public void Route(IEnumerable<IMeasurement> measurements)
	194	{
	195	lock (m_localCacheLock)
	196	{
	197	m_activeBuffer.Enqueue(measurements);
	198	}
	199	m_task.Start(1);
	200	}
	201
	202	public void Dispose()
	203	{
	204	m_task.Dispose();
	205	}
	206	}
	207	}
	208	}

Source/Libraries/GSF.TimeSeries/Adapters/RoutingTables.cs+9−3

     /// <summary>
     /// Represents the routing tables for the Iaon adapters.
     /// </summary>
-    public class RoutingTables : IDisposable
+    public partial class RoutingTables : IDisposable
     {
         #region [ Members ]
         private class GlobalCache
         {
             public HashSet<IAdapter> ProducerAdapters;
             public Dictionary<Guid, List<Consumer>> GlobalSignalLookup;
             public Dictionary<IAdapter, Consumer> GlobalDestinationLookup;
             public List<Consumer> BroadcastConsumers;
         private bool m_disposed;
+        private RoutingTableInputBuffer m_inputBuffer;
         #endregion
         #region [ Constructors ]
         public RoutingTables()
         {
             m_calculateRoutingTablesOperation = new LongSynchronizedOperation(CalculateRoutingTables) { IsBackground = true };
+            m_inputBuffer = new RoutingTableInputBuffer(this);
         }
         /// <summary>
                 {
                     if (disposing)
                     {
+                        m_inputBuffer.Dispose();
                         m_inputAdapters = null;
                         m_actionAdapters = null;
                         m_outputAdapters = null;
                 // Start or stop any connect on demand adapters
                 HandleConnectOnDemandAdapters(new HashSet<MeasurementKey>(m_inputMeasurementKeysRestriction ?? Enumerable.Empty<MeasurementKey>()), inputAdapterCollection, actionAdapterCollection, outputAdapterCollection);
                 elapsedTime = Ticks.ToSeconds(DateTime.UtcNow.Ticks - startTime);
                 routeCount = globalSignalLookup.Count;
         /// <returns>The measurement handler used for routing.</returns>
         public virtual EventHandler<EventArgs<ICollection<IMeasurement>>> GetRoutedMeasurementsHandler()
         {
+            return (sender, args) => m_inputBuffer.Route(args.Argument);
             LocalCache localCache = new LocalCache(this);
             return (sender, args) => localCache.Route(args.Argument);
         }

Source/Libraries/GSF.TimeSeries/GSF.TimeSeries.csproj+1−0

@@ -100,6 +100,7 @@ · 99 unmodified lines
100	<Compile Include="Adapters\ConnectionStringParameterAttribute.cs" />	100	<Compile Include="Adapters\ConnectionStringParameterAttribute.cs" />
101	<Compile Include="Adapters\CustomConfigurationEditorAttribute.cs" />	101	<Compile Include="Adapters\CustomConfigurationEditorAttribute.cs" />
102	<Compile Include="Adapters\IaonSession.cs" />	102	<Compile Include="Adapters\IaonSession.cs" />
		103	<Compile Include="Adapters\RoutingTableInputBuffer.cs" />
103	<Compile Include="Adapters\RoutingTables.cs" />	104	<Compile Include="Adapters\RoutingTables.cs" />
104	<Compile Include="Alarm.cs" />	105	<Compile Include="Alarm.cs" />
105	<Compile Include="ConcentratorBase.cs" />	106	<Compile Include="ConcentratorBase.cs" />

GSF.TimeSeries: Added a new routing method designed to process measurements in batch and reduce overall locking required.

Keyboard shortcuts

General

Issues and pull requests

Issues, pull requests and Inbox

Repository code