//****************************************************************************************************** // DoubleBufferedQueueManager.cs - Gbtc // // Copyright © 2014, 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: // ---------------------------------------------------------------------------------------------------- // 02/15/2014 - Stephen C. Wills // Generated original version of source code. // //****************************************************************************************************** using System; using System.Collections.Generic; using System.Threading; using GSF.Threading; namespace GSF.Collections { /// /// A producer for a which can /// only be used to provide items to the queue for consumption. /// /// The type of the items produced to the queue. public class DoubleBufferedQueueProducer : IDisposable { #region [ Members ] // Fields private DoubleBufferedQueueManager m_manager; private DoubleBufferedQueue m_queue; private bool m_disposed; #endregion #region [ Constructors ] /// /// Creates a new instance of the class. /// /// The that created this producer. /// The that this producer will be producing to. internal DoubleBufferedQueueProducer(DoubleBufferedQueueManager manager, DoubleBufferedQueue queue) { if ((object)manager == null) throw new ArgumentNullException(nameof(manager)); if ((object)queue == null) throw new ArgumentNullException(nameof(queue)); m_manager = manager; m_queue = queue; } /// /// Releases the unmanaged resources before the object is reclaimed by . /// ~DoubleBufferedQueueProducer() { Dispose(false); } #endregion #region [ Methods ] /// /// Produces a collection of items to be processed by the consumer. /// /// The collection of items to be enqueued. public void Produce(IEnumerable items) { DoubleBufferedQueue queue = m_queue; if ((object)queue != null) { m_queue.Enqueue(items); m_manager.SignalItemHandler(); } } /// /// Releases all the resources used by the object. /// public void Dispose() { Dispose(true); GC.SuppressFinalize(this); } /// /// Releases the unmanaged resources used by the object and optionally releases the managed resources. /// /// true to release both managed and unmanaged resources; false to release only unmanaged resources. protected virtual void Dispose(bool disposing) { if (!m_disposed) { try { DoubleBufferedQueue queue = Interlocked.Exchange(ref m_queue, null); if ((object)queue != null) m_manager.ReturnQueue(queue); } finally { m_disposed = true; // Prevent duplicate dispose. } } } #endregion } /// /// Manages queues to reduce contention for a multithreaded, multiple-producer, single-consumer scenario. /// /// /// For best results, each thread that is producing items to the consumer should call /// to receive a producer object that will not contend with /// any other producer. The consumer should either provide a handler to process the queued /// items or poll the manager by calling (not both!). It is not /// safe to use this class with multiple consumer threads. /// /// The types of items to be queued. /// /// It is not safe to use this class with multiple consumer threads. /// The list returned by is not thread-safe and /// is reused on each Dequeue operation, so no other thread should /// access the list while another thread is calling Dequeue. /// public class DoubleBufferedQueueManager { #region [ Members ] // Fields private readonly ShortSynchronizedOperation m_itemHandlingOperation; private readonly Action m_itemHandler; private readonly List> m_queues; private readonly List m_dequeuedItems; private volatile bool m_itemsLeft; private readonly object m_queuesLock; #endregion #region [ Constructors ] /// /// Creates a new instance of the class. /// public DoubleBufferedQueueManager() { m_queues = new List>(); m_dequeuedItems = new List(); m_queuesLock = new object(); } /// /// Creates a new instance of the class. /// /// The method to handle processing of queued items. public DoubleBufferedQueueManager(Action itemHandler) : this(itemHandler, null) { } /// /// Creates a new instance of the class. /// /// The method to handle processing of queued items. public DoubleBufferedQueueManager(Action> itemHandler) : this(itemHandler, null) { } /// /// Creates a new instance of the class. /// /// The method to handle processing of queued items. /// The method to handle exceptions that occur when processing items. public DoubleBufferedQueueManager(Action itemHandler, Action exceptionHandler) : this() { m_itemHandlingOperation = new ShortSynchronizedOperation(CallItemHandler, exceptionHandler); m_itemHandler = itemHandler; } /// /// Creates a new instance of the class. /// /// The method to handle processing of queued items. /// The method to handle exceptions that occur when processing items. public DoubleBufferedQueueManager(Action> itemHandler, Action exceptionHandler) : this() { m_itemHandlingOperation = new ShortSynchronizedOperation(CallItemHandler, exceptionHandler); m_itemHandler = () => { IList items = Dequeue(); if (items.Count > 0) itemHandler(items); }; } #endregion #region [ Properties ] /// /// Gets a flag that indicates whether there are any items left to /// be consumed after the last call to . /// public bool ItemsLeft { get { return m_itemsLeft; } } #endregion #region [ Methods ] /// /// Creates a producer used to produce items to the consumer of this . /// /// A used to produce items to the consumer. public DoubleBufferedQueueProducer GetProducer() { DoubleBufferedQueue queue = new DoubleBufferedQueue(); lock (m_queuesLock) { m_queues.Add(queue); } return new DoubleBufferedQueueProducer(this, queue); } /// /// Dequeues a list of items produced by the s. /// /// A list of items to be consumed. public IList Dequeue() { IList dequeuedItems; lock (m_queuesLock) { m_itemsLeft = false; if (m_queues.Count == 0) return EmptyList; m_dequeuedItems.Clear(); foreach (DoubleBufferedQueue queue in m_queues) { if (queue.TryDequeue(out dequeuedItems) <= 0) { m_dequeuedItems.AddRange(dequeuedItems); // Clearing the list is not necessary, but is a nice // optimization allowing the garbage collector to // potentially clean up the items before the next // dequeue and to reduce the amount of time spent in // the DoubleBufferedQueue's lock since it won't have // to clear the list during the next dequeue operation if (!dequeuedItems.IsReadOnly) dequeuedItems.Clear(); } else { m_itemsLeft = true; } } } return m_dequeuedItems; } /// /// Runs the operation to process items produced by the s. /// internal void SignalItemHandler() { if ((object)m_itemHandlingOperation != null) m_itemHandlingOperation.RunOnceAsync(); } /// /// Returns a queue to the /// so that it can be removed from the list of queues to be consumed. /// /// The queue to be returned. internal void ReturnQueue(DoubleBufferedQueue queue) { int last; int index; lock (m_queuesLock) { index = m_queues.IndexOf(queue); if (index >= 0) { last = m_queues.Count - 1; m_queues[index] = m_queues[last]; m_queues.RemoveAt(last); } } } private void CallItemHandler() { m_itemHandler(); if (m_itemsLeft) m_itemHandlingOperation.RunOnceAsync(); } #endregion #region [ Static ] // Static Fields private static readonly IList EmptyList = new T[0]; #endregion } }