//******************************************************************************************************//  ProcessList.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://  ----------------------------------------------------------------------------------------------------//  01/07/2006 - J. Ritchie Carroll//       Generated original version of source code.//  02/12/2006 - J. Ritchie Carroll//       Added multi-item bulk processing functionality.//  03/21/2007 - J. Ritchie Carroll//       Added "ItemsBeingProcessed" property to return current total number of items being processed.//       Added "Flush" method to allow any remaining items in list to be processed before shutdown.//  04/05/2007 - J. Ritchie Carroll//       Added "RequeueMode" properties to allow users to specify how data gets reinserted back into//       the list (prefix or suffix) after processing timeouts or exceptions.//  07/12/2007 - Pinal C. Patel//       Modified the code for "Flush" method to correctly implement IDisposable interface.//  08/01/2007 - J. Ritchie Carroll//       Added some minor optimizations where practical.//  08/17/2007 - J. Ritchie Carroll//       Removed IDisposable implementation because of continued flushing errors.//  08/17/2007 - Darrell Zuercher//       Edited code comments.//  11/05/2007 - J. Ritchie Carroll//       Modified flush to complete tasks on calling thread - this avoids errors when timer//       gets disposed before flush call.//  02/20/2008 - J. Ritchie Carroll//       Implemented standard IDisposable pattern.//  09/11/2008 - J. Ritchie Carroll//       Converted to C#.//  11/06/2008 - J. Ritchie Carroll//       Added CurrentStatistics property to return run-time statistics as a group.//  02/23/2009 - Josh L. Patterson//       Edited Code Comments.//  08/05/2009 - Josh L. Patterson//       Edited Code Comments.//  09/14/2009 - Stephen C. Wills//       Added new header and license agreement.//  01/04/2010 - J. Ritchie Carroll//       Removed hard thread abort from shutdown which will allow current processing items//       to complete before terminating thread.//  06/21/2010 - Stephen C. Wills//       Modified Dispose to fix potential concurrency issues.//  11/23/2011 - J. Ritchie Carroll//       Modified to support buffer optimized ISupportBinaryImage.//  01/24/2012 - Pinal C. Patel//       Modified ProcessTimerThreadProc() method to perform null reference check to avoid an exception//       when the object is being disposed.//  11/27/2012 - J. Ritchie Carroll//       Modified process queue to be based on ConcurrentQueue<T>. New ProcessList<T> class now assumes//       the List<T> based functionality of original code.//  12/13/2012 - Starlynn Danyelle Gilliam//       Modified Header.//  04/03/2014 - J. Ritchie Carroll//       Changed to List<T> with sync-lock operation since this is faster than ConcurrentQueue<T>//       when dealing with collections of items. ProcessQueue was dropped in lieu of ProcessList.////****************************************************************************************************** using System;using System.Collections;using System.Collections.Generic;using System.Diagnostics.CodeAnalysis;using System.Linq;using System.Text;using System.Threading;using System.Threading.Tasks;using System.Timers;using GSF.Threading;using GSF.Units;using Timer = System.Timers.Timer; namespace GSF.Collections{    #region [ Enumerations ]     /// <summary>    /// Enumeration of possible <see cref="ProcessQueue{T}"/> threading modes.    /// </summary>    public enum QueueThreadingMode    {        /// <summary>        /// Processes several items in the <see cref="ProcessQueue{T}"/> at once on different threads, where processing order is not important.        /// </summary>        Asynchronous,        /// <summary>        /// Processes items in the <see cref="ProcessQueue{T}"/> one at a time on a single thread, where processing order is important.        /// </summary>        Synchronous    }     /// <summary>    /// Enumeration of possible <see cref="ProcessQueue{T}"/> processing styles.    /// </summary>    public enum QueueProcessingStyle    {        /// <summary>        /// Defines <see cref="ProcessQueue{T}"/> processing delegate to process only one item at a time.        /// </summary>        /// <remarks>        /// This is the typical <see cref="QueueProcessingStyle"/> when the <see cref="QueueThreadingMode"/> is asynchronous.        /// </remarks>        OneAtATime,        /// <summary>        /// Defines <see cref="ProcessQueue{T}"/> processing delegate to process all currently available items in the <see cref="ProcessQueue{T}"/>.        /// Items are passed into delegate as an array.        /// </summary>        /// <remarks>        /// This is the optimal <see cref="QueueProcessingStyle"/> when the <see cref="QueueThreadingMode"/> is synchronous.        /// </remarks>        ManyAtOnce    }     /// <summary>    /// Enumeration of possible requeue reasons.    /// </summary>    public enum RequeueReason    {        /// <summary>        /// Requeuing item since it cannot be processed at this time.        /// </summary>        CannotProcess,        /// <summary>        /// Requeuing item due to an exception.        /// </summary>        Exception,        /// <summary>        /// Requeing item due to timeout.        /// </summary>        Timeout    }     /// <summary>    /// Enumeration of possible requeue modes.    /// </summary>    public enum RequeueMode    {        /// <summary>        /// Requeues item at the beginning of the <see cref="ProcessQueue{T}"/>.        /// </summary>        Prefix,        /// <summary>        /// Requeues item at the end of the <see cref="ProcessQueue{T}"/>.        /// </summary>        Suffix    }     #endregion     /// <summary>    /// Represents a thread-safe (via locking) list of items, based on <see cref="List{T}"/>, that get processed on independent threads with a consumer provided function.    /// </summary>    /// <typeparam name="T">Type of object to process</typeparam>    /// <remarks>    /// <para>This class acts as a strongly-typed collection of objects to be processed.</para>    /// <para>Note that the <see cref="ProcessQueue{T}"/> will not start processing until the Start method is called.</para>    /// </remarks>    public class ProcessQueue<T> : IList<T>, IProvideStatus, ISupportLifecycle    {        #region [ Members ]         // Nested Types         // Limits item processing time, if requested.        private sealed class TemporalTask : IDisposable        {            private readonly ProcessQueue<T> m_parent;            private readonly Task m_task;            private readonly T m_item;            private readonly T[] m_items;            private bool m_disposed;             private TemporalTask(ProcessQueue<T> parent, T item)            {                m_parent = parent;                m_item = item;                Thread.MemoryBarrier();                m_task = Task.Factory.StartNew(ProcessItem);            }             private TemporalTask(ProcessQueue<T> parent, T[] items)            {                m_parent = parent;                m_items = items;                Thread.MemoryBarrier();                m_task = Task.Factory.StartNew(ProcessItems);            }             ~TemporalTask()            {                Dispose(false);            }             private void ProcessItem()            {                // In rare cases, this method was throwing a NullReferenceException which seems                // to indicate that m_parent is null. At first glance, this should not be possible                // as m_parent is readonly and the constructor is invoked by m_parent itself.                // However, note that m_task is also assigned in the constructor by calling                // Task.Factory.StartNew, which executes this method on a separate thread. If we                // instead assume that the assignment operations in the constructor could be                // reordered, or if this method executes on a separate processor before the first                // processor's cache is flushed, then it should be possible for m_parent and even                // m_item to be null. The memory barriers should prevent this reordering of events.                Thread.MemoryBarrier();                m_parent.ProcessItem(m_item);            }             private void ProcessItems()            {                // In rare cases, this method was throwing a NullReferenceException which seems                // to indicate that m_parent is null. At first glance, this should not be possible                // as m_parent is readonly and the constructor is invoked by m_parent itself.                // However, note that m_task is also assigned in the constructor by calling                // Task.Factory.StartNew, which executes this method on a separate thread. If we                // instead assume that the assignment operations in the constructor could be                // reordered, or if this method executes on a separate processor before the first                // processor's cache is flushed, then it should be possible for m_parent and even                // m_items to be null. The memory barriers should prevent this reordering of events.                Thread.MemoryBarrier();                m_parent.ProcessItems(m_items);            }             // Blocks calling thread until specified process timeout has expired.            private bool Wait()            {                return m_task.Wait(m_parent.ProcessTimeout);            }             void IDisposable.Dispose()            {                Dispose(true);                GC.SuppressFinalize(this);            }             private void Dispose(bool disposing)            {                if (!m_disposed)                {                    try                    {                        if (disposing)                        {                            if ((object)m_task != null && (m_task.Status == TaskStatus.RanToCompletion || m_task.Status == TaskStatus.Faulted || m_task.Status == TaskStatus.Canceled))                                m_task.Dispose();                        }                    }                    finally                    {                        m_disposed = true;  // Prevent duplicate dispose.                    }                }            }             public static bool Process(ProcessQueue<T> parent, T item)            {                using (TemporalTask temporalProcess = new TemporalTask(parent, item))                {                    return temporalProcess.Wait();                }            }             public static bool Process(ProcessQueue<T> parent, T[] items)            {                using (TemporalTask temporalProcess = new TemporalTask(parent, items))                {                    return temporalProcess.Wait();                }            }        }         // Constants         /// <summary>        /// Default processing interval (in milliseconds).        /// </summary>        public const int DefaultProcessInterval = 100;         /// <summary>        /// Default maximum number of processing threads.        /// </summary>        public const int DefaultMaximumThreads = 5;         /// <summary>        /// Default processing timeout (in milliseconds).        /// </summary>        public const int DefaultProcessTimeout = Timeout.Infinite;         /// <summary>        /// Default setting for requeuing items on processing timeout.        /// </summary>        public const bool DefaultRequeueOnTimeout = false;         /// <summary>        /// Default setting for requeuing items on processing exceptions.        /// </summary>        public const bool DefaultRequeueOnException = false;         /// <summary>        /// Default real-time processing interval (in milliseconds).        /// </summary>        public const double RealTimeProcessInterval = 0.0;         /// <summary>        /// Default setting for requeuing mode on processing timeout.        /// </summary>        public const RequeueMode DefaultRequeueModeOnTimeout = RequeueMode.Prefix;         /// <summary>        /// Default setting for requeuing mode on processing exceptions.        /// </summary>        public const RequeueMode DefaultRequeueModeOnException = RequeueMode.Prefix;         // Delegates         /// <summary>        /// Function signature that defines a method to process items one at a time.        /// </summary>        /// <param name="item">Item to be processed.</param>        /// <remarks>        /// <para>Required unless <see cref="ProcessQueue{T}.ProcessItemsFunction"/> is implemented.</para>        /// <para>Creates an asynchronous <see cref="ProcessQueue{T}"/> to process individual items - one item at a time - on multiple threads.</para>        /// </remarks>        [SuppressMessage("Microsoft.Design", "CA1034:NestedTypesShouldNotBeVisible")]        public delegate void ProcessItemFunctionSignature(T item);         /// <summary>        /// Function signature that defines a method to process multiple items at once.        /// </summary>        /// <param name="items">Items to be processed.</param>        /// <remarks>        /// <para>Required unless <see cref="ProcessQueue{T}.ProcessItemFunction"/> is implemented.</para>        /// <para>Creates an asynchronous <see cref="ProcessQueue{T}"/> to process groups of items simultaneously on multiple threads.</para>        /// </remarks>        [SuppressMessage("Microsoft.Design", "CA1034:NestedTypesShouldNotBeVisible")]        public delegate void ProcessItemsFunctionSignature(T[] items);         /// <summary>        /// Function signature that determines if an item can be currently processed.        /// </summary>        /// <param name="item">Item to be checked for processing availability.</param>        /// <returns>True, if item can be processed. The default is true.</returns>        /// <remarks>        /// <para>Implementation of this function is optional. It is assumed that an item can be processed if this        /// function is not defined</para>        /// <para>Items must eventually get to a state where they can be processed, or they will remain in the <see cref="ProcessQueue{T}"/>        /// indefinitely.</para>        /// <para>        /// Note that when this function is implemented and ProcessingStyle = ManyAtOnce (i.e.,         /// <see cref="ProcessQueue{T}.ProcessItemsFunction"/> is defined), then each item presented for         /// processing must evaluate as "CanProcessItem = True" before any items are processed.        /// </para>        /// </remarks>        [SuppressMessage("Microsoft.Design", "CA1034:NestedTypesShouldNotBeVisible")]        public delegate bool CanProcessItemFunctionSignature(T item);         // Events         /// <summary>        /// Event that is raised after an item has been successfully processed.        /// </summary>        /// <remarks>        /// <para>Allows custom handling of successfully processed items.</para>        /// <para>Allows notification when an item has completed processing in the allowed amount of time, if a process        /// timeout is specified.</para>        /// <para>Raised only when ProcessingStyle = OneAtATime (i.e., <see cref="ProcessQueue{T}.ProcessItemFunction"/> is defined).</para>        /// </remarks>        public event EventHandler<EventArgs<T>> ItemProcessed;         /// <summary>        /// Event that is raised after an array of items have been successfully processed.        /// </summary>        /// <remarks>        /// <para>Allows custom handling of successfully processed items.</para>        /// <para>Allows notification when an item has completed processing in the allowed amount of time, if a process        /// timeout is specified.</para>        /// <para>Raised only when when ProcessingStyle = ManyAtOnce (i.e., <see cref="ProcessQueue{T}.ProcessItemsFunction"/> is defined).</para>        /// </remarks>        public event EventHandler<EventArgs<T[]>> ItemsProcessed;         /// <summary>        /// Event that is raised if an item's processing time exceeds the specified process timeout.        /// </summary>        /// <remarks>        /// <para>Allows custom handling of items that took too long to process.</para>        /// <para>Raised only when ProcessingStyle = OneAtATime (i.e., <see cref="ProcessQueue{T}.ProcessItemFunction"/> is defined).</para>        /// </remarks>        public event EventHandler<EventArgs<T>> ItemTimedOut;         /// <summary>        /// Event that is raised if the processing time for an array of items exceeds the specified process timeout.        /// </summary>        /// <remarks>        /// <para>Allows custom handling of items that took too long to process.</para>        /// <para>Raised only when ProcessingStyle = ManyAtOnce (i.e., <see cref="ProcessQueue{T}.ProcessItemsFunction"/> is defined).</para>        /// </remarks>        public event EventHandler<EventArgs<T[]>> ItemsTimedOut;         /// <summary>        /// Event that is raised if an exception is encountered while attempting to processing an item in the <see cref="ProcessQueue{T}"/>.        /// </summary>        /// <remarks>        /// Processing will not stop for any exceptions thrown by the user function, but any captured exceptions will        /// be exposed through this event.        /// </remarks>        public event EventHandler<EventArgs<Exception>> ProcessException;         /// <summary>        /// Occurs when the class has been disposed.        /// </summary>        public event EventHandler Disposed;         // Fields        private ProcessItemFunctionSignature m_processItemFunction;        private ProcessItemsFunctionSignature m_processItemsFunction;        private CanProcessItemFunctionSignature m_canProcessItemFunction;         private ISynchronizedOperation m_synchronizedOperation;        private SynchronizedOperationType m_synchronizedOperationType;         private IList<T> m_processList;        private int m_maximumThreads;        private int m_processTimeout;        private readonly bool m_processingIsRealTime;         private bool m_requeueOnTimeout;        private bool m_requeueOnException;        private RequeueMode m_requeueModeOnTimeout;        private RequeueMode m_requeueModeOnException;         private volatile bool m_enabled;        private int m_threadCount;        private long m_itemsBeingProcessed;        private long m_totalProcessedItems;        private long m_totalFunctionCalls;        private long m_startTime;        private long m_stopTime;        private string m_name;        private bool m_disposed;         private Timer m_processTimer;         #endregion         #region [ Constructors ]         /// <summary>        /// Creates a <see cref="ProcessQueue{T}"/> based on the generic List(Of T) class.        /// </summary>        /// <param name="processItemFunction">Delegate that defines a method to process one item at a time.</param>        /// <param name="processInterval">a <see cref="double"/> value which represents the process interval in milliseconds.</param>        /// <param name="maximumThreads">The maximum number of threads for the queue to use.</param>        /// <param name="processTimeout">The number of seconds before a process should timeout.</param>        /// <param name="requeueOnTimeout">A <see cref="Boolean"/> value that indicates whether a process should requeue an item on timeout.</param>        /// <param name="requeueOnException">A <see cref="Boolean"/> value that indicates whether a process should requeue after an exception.</param>        public ProcessQueue(ProcessItemFunctionSignature processItemFunction, double processInterval = DefaultProcessInterval, int maximumThreads = DefaultMaximumThreads, int processTimeout = DefaultProcessTimeout, bool requeueOnTimeout = DefaultRequeueOnTimeout, bool requeueOnException = DefaultRequeueOnException)            : this(processItemFunction, null, null, new List<T>(), processInterval, maximumThreads, processTimeout, requeueOnTimeout, requeueOnException)        {        }         /// <summary>        /// Creates a <see cref="ProcessQueue{T}"/> based on the generic List(Of T) class.        /// </summary>        /// <param name="processItemFunction">Delegate that defines a method to process one item at a time.</param>        /// <param name="canProcessItemFunction">Delegate that determines if an item can currently be processed.</param>        /// <param name="processInterval">a <see cref="double"/> value which represents the process interval in milliseconds.</param>        /// <param name="maximumThreads">The maximum number of threads for the queue to use.</param>        /// <param name="processTimeout">The number of seconds before a process should timeout.</param>        /// <param name="requeueOnTimeout">A <see cref="Boolean"/> value that indicates whether a process should requeue an item on timeout.</param>        /// <param name="requeueOnException">A <see cref="Boolean"/> value that indicates whether a process should requeue after an exception.</param>        public ProcessQueue(ProcessItemFunctionSignature processItemFunction, CanProcessItemFunctionSignature canProcessItemFunction = null, double processInterval = DefaultProcessInterval, int maximumThreads = DefaultMaximumThreads, int processTimeout = DefaultProcessTimeout, bool requeueOnTimeout = DefaultRequeueOnTimeout, bool requeueOnException = DefaultRequeueOnException)            : this(processItemFunction, null, canProcessItemFunction, new List<T>(), processInterval, maximumThreads, processTimeout, requeueOnTimeout, requeueOnException)        {        }         /// <summary>        /// Creates a bulk item <see cref="ProcessQueue{T}"/> based on the generic List(Of T) class.        /// </summary>        /// <param name="processItemsFunction">Delegate that defines a method to process multiple items at once.</param>        /// <param name="processInterval">a <see cref="double"/> value which represents the process interval in milliseconds.</param>        /// <param name="maximumThreads">The maximum number of threads for the queue to use.</param>        /// <param name="processTimeout">The number of seconds before a process should timeout.</param>        /// <param name="requeueOnTimeout">A <see cref="Boolean"/> value that indicates whether a process should requeue an item on timeout.</param>        /// <param name="requeueOnException">A <see cref="Boolean"/> value that indicates whether a process should requeue after an exception.</param>        public ProcessQueue(ProcessItemsFunctionSignature processItemsFunction, double processInterval = DefaultProcessInterval, int maximumThreads = DefaultMaximumThreads, int processTimeout = DefaultProcessTimeout, bool requeueOnTimeout = DefaultRequeueOnTimeout, bool requeueOnException = DefaultRequeueOnException)            : this(null, processItemsFunction, null, new List<T>(), processInterval, maximumThreads, processTimeout, requeueOnTimeout, requeueOnException)        {

Lines 1–500 of 3426. Load more