//******************************************************************************************************* // ProcessQueue.cs - Gbtc // // Tennessee Valley Authority, 2011 // No copyright is claimed pursuant to 17 USC § 105. All Other Rights Reserved. // // This software is made freely available under the TVA Open Source Agreement (see below). // Code in this file licensed to GSF under one or more contributor license agreements listed below. // // 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 queue 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.. // //******************************************************************************************************* #region [ TVA Open Source Agreement ] /* THIS OPEN SOURCE AGREEMENT ("AGREEMENT") DEFINES THE RIGHTS OF USE,REPRODUCTION, DISTRIBUTION, MODIFICATION AND REDISTRIBUTION OF CERTAIN COMPUTER SOFTWARE ORIGINALLY RELEASED BY THE TENNESSEE VALLEY AUTHORITY, A CORPORATE AGENCY AND INSTRUMENTALITY OF THE UNITED STATES GOVERNMENT ("GOVERNMENT AGENCY"). GOVERNMENT AGENCY IS AN INTENDED THIRD-PARTY BENEFICIARY OF ALL SUBSEQUENT DISTRIBUTIONS OR REDISTRIBUTIONS OF THE SUBJECT SOFTWARE. ANYONE WHO USES, REPRODUCES, DISTRIBUTES, MODIFIES OR REDISTRIBUTES THE SUBJECT SOFTWARE, AS DEFINED HEREIN, OR ANY PART THEREOF, IS, BY THAT ACTION, ACCEPTING IN FULL THE RESPONSIBILITIES AND OBLIGATIONS CONTAINED IN THIS AGREEMENT. Original Software Designation: openPDC Original Software Title: The GSF Open Source Phasor Data Concentrator User Registration Requested. Please Visit https://naspi.tva.com/Registration/ Point of Contact for Original Software: J. Ritchie Carroll 1. DEFINITIONS A. "Contributor" means Government Agency, as the developer of the Original Software, and any entity that makes a Modification. B. "Covered Patents" mean patent claims licensable by a Contributor that are necessarily infringed by the use or sale of its Modification alone or when combined with the Subject Software. C. "Display" means the showing of a copy of the Subject Software, either directly or by means of an image, or any other device. D. "Distribution" means conveyance or transfer of the Subject Software, regardless of means, to another. E. "Larger Work" means computer software that combines Subject Software, or portions thereof, with software separate from the Subject Software that is not governed by the terms of this Agreement. F. "Modification" means any alteration of, including addition to or deletion from, the substance or structure of either the Original Software or Subject Software, and includes derivative works, as that term is defined in the Copyright Statute, 17 USC § 101. However, the act of including Subject Software as part of a Larger Work does not in and of itself constitute a Modification. G. "Original Software" means the computer software first released under this Agreement by Government Agency entitled openPDC, including source code, object code and accompanying documentation, if any. H. "Recipient" means anyone who acquires the Subject Software under this Agreement, including all Contributors. I. "Redistribution" means Distribution of the Subject Software after a Modification has been made. J. "Reproduction" means the making of a counterpart, image or copy of the Subject Software. K. "Sale" means the exchange of the Subject Software for money or equivalent value. L. "Subject Software" means the Original Software, Modifications, or any respective parts thereof. M. "Use" means the application or employment of the Subject Software for any purpose. 2. GRANT OF RIGHTS A. Under Non-Patent Rights: Subject to the terms and conditions of this Agreement, each Contributor, with respect to its own contribution to the Subject Software, hereby grants to each Recipient a non-exclusive, world-wide, royalty-free license to engage in the following activities pertaining to the Subject Software: 1. Use 2. Distribution 3. Reproduction 4. Modification 5. Redistribution 6. Display B. Under Patent Rights: Subject to the terms and conditions of this Agreement, each Contributor, with respect to its own contribution to the Subject Software, hereby grants to each Recipient under Covered Patents a non-exclusive, world-wide, royalty-free license to engage in the following activities pertaining to the Subject Software: 1. Use 2. Distribution 3. Reproduction 4. Sale 5. Offer for Sale C. The rights granted under Paragraph B. also apply to the combination of a Contributor's Modification and the Subject Software if, at the time the Modification is added by the Contributor, the addition of such Modification causes the combination to be covered by the Covered Patents. It does not apply to any other combinations that include a Modification. D. The rights granted in Paragraphs A. and B. allow the Recipient to sublicense those same rights. Such sublicense must be under the same terms and conditions of this Agreement. 3. OBLIGATIONS OF RECIPIENT A. Distribution or Redistribution of the Subject Software must be made under this Agreement except for additions covered under paragraph 3H. 1. Whenever a Recipient distributes or redistributes the Subject Software, a copy of this Agreement must be included with each copy of the Subject Software; and 2. If Recipient distributes or redistributes the Subject Software in any form other than source code, Recipient must also make the source code freely available, and must provide with each copy of the Subject Software information on how to obtain the source code in a reasonable manner on or through a medium customarily used for software exchange. B. Each Recipient must ensure that the following copyright notice appears prominently in the Subject Software: No copyright is claimed pursuant to 17 USC § 105. All Other Rights Reserved. C. Each Contributor must characterize its alteration of the Subject Software as a Modification and must identify itself as the originator of its Modification in a manner that reasonably allows subsequent Recipients to identify the originator of the Modification. In fulfillment of these requirements, Contributor must include a file (e.g., a change log file) that describes the alterations made and the date of the alterations, identifies Contributor as originator of the alterations, and consents to characterization of the alterations as a Modification, for example, by including a statement that the Modification is derived, directly or indirectly, from Original Software provided by Government Agency. Once consent is granted, it may not thereafter be revoked. D. A Contributor may add its own copyright notice to the Subject Software. Once a copyright notice has been added to the Subject Software, a Recipient may not remove it without the express permission of the Contributor who added the notice. E. A Recipient may not make any representation in the Subject Software or in any promotional, advertising or other material that may be construed as an endorsement by Government Agency or by any prior Recipient of any product or service provided by Recipient, or that may seek to obtain commercial advantage by the fact of Government Agency's or a prior Recipient's participation in this Agreement. F. In an effort to track usage and maintain accurate records of the Subject Software, each Recipient, upon receipt of the Subject Software, is requested to register with Government Agency by visiting the following website: https://naspi.tva.com/Registration/. Recipient's name and personal information shall be used for statistical purposes only. Once a Recipient makes a Modification available, it is requested that the Recipient inform Government Agency at the web site provided above how to access the Modification. G. Each Contributor represents that that its Modification does not violate any existing agreements, regulations, statutes or rules, and further that Contributor has sufficient rights to grant the rights conveyed by this Agreement. H. A Recipient may choose to offer, and to charge a fee for, warranty, support, indemnity and/or liability obligations to one or more other Recipients of the Subject Software. A Recipient may do so, however, only on its own behalf and not on behalf of Government Agency or any other Recipient. Such a Recipient must make it absolutely clear that any such warranty, support, indemnity and/or liability obligation is offered by that Recipient alone. Further, such Recipient agrees to indemnify Government Agency and every other Recipient for any liability incurred by them as a result of warranty, support, indemnity and/or liability offered by such Recipient. I. A Recipient may create a Larger Work by combining Subject Software with separate software not governed by the terms of this agreement and distribute the Larger Work as a single product. In such case, the Recipient must make sure Subject Software, or portions thereof, included in the Larger Work is subject to this Agreement. J. Notwithstanding any provisions contained herein, Recipient is hereby put on notice that export of any goods or technical data from the United States may require some form of export license from the U.S. Government. Failure to obtain necessary export licenses may result in criminal liability under U.S. laws. Government Agency neither represents that a license shall not be required nor that, if required, it shall be issued. Nothing granted herein provides any such export license. 4. DISCLAIMER OF WARRANTIES AND LIABILITIES; WAIVER AND INDEMNIFICATION A. No Warranty: THE SUBJECT SOFTWARE IS PROVIDED "AS IS" WITHOUT ANY WARRANTY OF ANY KIND, EITHER EXPRESSED, IMPLIED, OR STATUTORY, INCLUDING, BUT NOT LIMITED TO, ANY WARRANTY THAT THE SUBJECT SOFTWARE WILL CONFORM TO SPECIFICATIONS, ANY IMPLIED WARRANTIES OF MERCHANTABILITY, FITNESS FOR A PARTICULAR PURPOSE, OR FREEDOM FROM INFRINGEMENT, ANY WARRANTY THAT THE SUBJECT SOFTWARE WILL BE ERROR FREE, OR ANY WARRANTY THAT DOCUMENTATION, IF PROVIDED, WILL CONFORM TO THE SUBJECT SOFTWARE. THIS AGREEMENT DOES NOT, IN ANY MANNER, CONSTITUTE AN ENDORSEMENT BY GOVERNMENT AGENCY OR ANY PRIOR RECIPIENT OF ANY RESULTS, RESULTING DESIGNS, HARDWARE, SOFTWARE PRODUCTS OR ANY OTHER APPLICATIONS RESULTING FROM USE OF THE SUBJECT SOFTWARE. FURTHER, GOVERNMENT AGENCY DISCLAIMS ALL WARRANTIES AND LIABILITIES REGARDING THIRD-PARTY SOFTWARE, IF PRESENT IN THE ORIGINAL SOFTWARE, AND DISTRIBUTES IT "AS IS." B. Waiver and Indemnity: RECIPIENT AGREES TO WAIVE ANY AND ALL CLAIMS AGAINST GOVERNMENT AGENCY, ITS AGENTS, EMPLOYEES, CONTRACTORS AND SUBCONTRACTORS, AS WELL AS ANY PRIOR RECIPIENT. IF RECIPIENT'S USE OF THE SUBJECT SOFTWARE RESULTS IN ANY LIABILITIES, DEMANDS, DAMAGES, EXPENSES OR LOSSES ARISING FROM SUCH USE, INCLUDING ANY DAMAGES FROM PRODUCTS BASED ON, OR RESULTING FROM, RECIPIENT'S USE OF THE SUBJECT SOFTWARE, RECIPIENT SHALL INDEMNIFY AND HOLD HARMLESS GOVERNMENT AGENCY, ITS AGENTS, EMPLOYEES, CONTRACTORS AND SUBCONTRACTORS, AS WELL AS ANY PRIOR RECIPIENT, TO THE EXTENT PERMITTED BY LAW. THE FOREGOING RELEASE AND INDEMNIFICATION SHALL APPLY EVEN IF THE LIABILITIES, DEMANDS, DAMAGES, EXPENSES OR LOSSES ARE CAUSED, OCCASIONED, OR CONTRIBUTED TO BY THE NEGLIGENCE, SOLE OR CONCURRENT, OF GOVERNMENT AGENCY OR ANY PRIOR RECIPIENT. RECIPIENT'S SOLE REMEDY FOR ANY SUCH MATTER SHALL BE THE IMMEDIATE, UNILATERAL TERMINATION OF THIS AGREEMENT. 5. GENERAL TERMS A. Termination: This Agreement and the rights granted hereunder will terminate automatically if a Recipient fails to comply with these terms and conditions, and fails to cure such noncompliance within thirty (30) days of becoming aware of such noncompliance. Upon termination, a Recipient agrees to immediately cease use and distribution of the Subject Software. All sublicenses to the Subject Software properly granted by the breaching Recipient shall survive any such termination of this Agreement. B. Severability: If any provision of this Agreement is invalid or unenforceable under applicable law, it shall not affect the validity or enforceability of the remainder of the terms of this Agreement. C. Applicable Law: This Agreement shall be subject to United States federal law only for all purposes, including, but not limited to, determining the validity of this Agreement, the meaning of its provisions and the rights, obligations and remedies of the parties. D. Entire Understanding: This Agreement constitutes the entire understanding and agreement of the parties relating to release of the Subject Software and may not be superseded, modified or amended except by further written agreement duly executed by the parties. E. Binding Authority: By accepting and using the Subject Software under this Agreement, a Recipient affirms its authority to bind the Recipient to all terms and conditions of this Agreement and that Recipient hereby agrees to all terms and conditions herein. F. Point of Contact: Any Recipient contact with Government Agency is to be directed to the designated representative as follows: J. Ritchie Carroll . */ #endregion #region [ Contributor License Agreements ] //****************************************************************************************************** // // Copyright © 2011, Grid Protection Alliance. All Rights Reserved. // // 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. // //****************************************************************************************************** #endregion using GSF.Units; using System; using System.Collections; using System.Collections.Concurrent; using System.Collections.Generic; using System.Linq; using System.Text; using System.Threading; namespace GSF.Collections { #region [ Enumerations ] /// /// Enumeration of possible threading modes. /// public enum QueueThreadingMode { /// /// Processes several items in the at once on different threads, where processing order is not important. /// Asynchronous, /// /// Processes items in the one at a time on a single thread, where processing order is important. /// Synchronous } /// /// Enumeration of possible processing styles. /// public enum QueueProcessingStyle { /// /// Defines processing delegate to process only one item at a time. /// /// /// This is the typical when the is asynchronous. /// OneAtATime, /// /// Defines processing delegate to process all currently available items in the . /// Items are passed into delegate as an array. /// /// /// This is the optimal when the is synchronous. /// ManyAtOnce } /// /// Enumeration of possible requeue reasons. /// public enum RequeueReason { /// /// Requeuing item since it cannot be processed at this time. /// CannotProcess, /// /// Requeuing item due to an exception. /// Exception, /// /// Requeing item due to timeout. /// Timeout } #endregion /// /// Represents a lock-free thread-safe collection of items, based on , that get processed on independent threads with a consumer provided function. /// /// Type of object to process /// /// This class acts as a strongly-typed collection of objects to be processed. /// Consumers are expected to create new instances of this class through the static construction functions (e.g., CreateAsynchronousQueue, CreateSynchronousQueue, etc.) /// Note that the will not start processing until the method is called. /// public class ProcessQueue : IProducerConsumerCollection, IProvideStatus, ISupportLifecycle { #region [ Members ] // Nested Types // Limits item processing time, if requested. private sealed class TemporalTask : IDisposable { private ProcessQueue m_parent; private ManualResetEventSlim m_eventHandle; private T m_item; private T[] m_items; private bool m_disposed; private TemporalTask(ProcessQueue parent, T item) { m_parent = parent; m_item = item; m_eventHandle = new ManualResetEventSlim(false); ThreadPool.QueueUserWorkItem(state => ProcessItem()); } private TemporalTask(ProcessQueue parent, T[] items) { m_parent = parent; m_items = items; m_eventHandle = new ManualResetEventSlim(false); ThreadPool.QueueUserWorkItem(state => ProcessItems()); } ~TemporalTask() { Dispose(false); } private void ProcessItem() { m_parent.ProcessItem(m_item); m_eventHandle.Set(); } private void ProcessItems() { m_parent.ProcessItems(m_items); m_eventHandle.Set(); } // Blocks calling thread until specified process timeout has expired. private bool Wait() { return m_eventHandle.Wait(m_parent.ProcessTimeout); } void IDisposable.Dispose() { Dispose(true); GC.SuppressFinalize(this); } private void Dispose(bool disposing) { if (!m_disposed) { try { if (disposing) { if (m_eventHandle != null) m_eventHandle.Dispose(); m_eventHandle = null; m_parent = null; } } finally { m_disposed = true; // Prevent duplicate dispose. } } } public static bool Process(ProcessQueue parent, T item) { using (TemporalTask temporalProcess = new TemporalTask(parent, item)) { return temporalProcess.Wait(); } } public static bool Process(ProcessQueue parent, T[] items) { using (TemporalTask temporalProcess = new TemporalTask(parent, items)) { return temporalProcess.Wait(); } } } // Constants /// Default processing interval (in milliseconds). public const int DefaultProcessInterval = 100; /// Default maximum number of processing threads. public const int DefaultMaximumThreads = 5; /// Default processing timeout (in milliseconds). public const int DefaultProcessTimeout = Timeout.Infinite; /// Default setting for requeuing items on processing timeout. public const bool DefaultRequeueOnTimeout = false; /// Default setting for requeuing items on processing exceptions. public const bool DefaultRequeueOnException = false; /// Default real-time processing interval (in milliseconds). public const double RealTimeProcessInterval = 0.0; // Delegates /// /// Function signature that defines a method to process items one at a time. /// /// Item to be processed. /// /// Required unless is implemented. /// Creates an asynchronous to process individual items - one item at a time - on multiple threads. /// public delegate void ProcessItemFunctionSignature(T item); /// /// Function signature that defines a method to process multiple items at once. /// /// Items to be processed. /// /// Required unless is implemented. /// Creates an asynchronous to process groups of items simultaneously on multiple threads. /// public delegate void ProcessItemsFunctionSignature(T[] items); /// /// Function signature that determines if an item can be currently processed. /// /// Item to be checked for processing availablity. /// True, if item can be processed. The default is true. /// /// Implementation of this function is optional. It is assumed that an item can be processed if this /// function is not defined /// Items must eventually get to a state where they can be processed, or they will remain in the /// indefinitely. /// /// Note that when this function is implemented and ProcessingStyle = ManyAtOnce (i.e., /// is defined), then each item presented for /// processing must evaluate as "CanProcessItem = True" before any items are processed. /// /// public delegate bool CanProcessItemFunctionSignature(T item); // Events /// /// Event that is raised after an item has been successfully processed. /// /// /// Allows custom handling of successfully processed items. /// Allows notification when an item has completed processing in the allowed amount of time, if a process /// timeout is specified. /// Raised only when ProcessingStyle = OneAtATime (i.e., is defined). /// public event EventHandler> ItemProcessed; /// /// Event that is raised after an array of items have been successfully processed. /// /// /// Allows custom handling of successfully processed items. /// Allows notification when an item has completed processing in the allowed amount of time, if a process /// timeout is specified. /// Raised only when when ProcessingStyle = ManyAtOnce (i.e., is defined). /// public event EventHandler> ItemsProcessed; /// /// Event that is raised if an item's processing time exceeds the specified process timeout. /// /// /// Allows custom handling of items that took too long to process. /// Raised only when ProcessingStyle = OneAtATime (i.e., is defined). /// public event EventHandler> ItemTimedOut; /// /// Event that is raised if the processing time for an array of items exceeds the specified process timeout. /// /// /// Allows custom handling of items that took too long to process. /// Raised only when ProcessingStyle = ManyAtOnce (i.e., is defined). /// public event EventHandler> ItemsTimedOut; /// /// Event that is raised if an exception is encountered while attempting to processing an item in the . /// /// /// Processing will not stop for any exceptions thrown by the user function, but any captured exceptions will /// be exposed through this event. /// public event EventHandler> ProcessException; /// /// Occurs when the class has been disposed. /// public event EventHandler Disposed; // Fields private ProcessItemFunctionSignature m_processItemFunction; private ProcessItemsFunctionSignature m_processItemsFunction; private CanProcessItemFunctionSignature m_canProcessItemFunction; private IEnumerable m_processQueue; private int m_maximumThreads; private int m_processTimeout; private readonly bool m_processingIsRealTime; private bool m_requeueOnTimeout; private bool m_requeueOnException; private int m_processing; private int m_threadCount; private bool m_enabled; private long m_itemsProcessing; private long m_itemsProcessed; private long m_startTime; private long m_stopTime; private string m_name; private bool m_disposed; #if ThreadTracking private ManagedThread m_realTimeProcessThread; #else private Thread m_realTimeProcessThread; #endif private System.Timers.Timer m_processTimer; #endregion #region [ Constructors ] /// /// Creates a based on the generic List(Of T) class. /// /// Delegate that defines a method to process one item at a time. /// Delegate that determines if an item can currently be processed. /// a value which represents the process interval in milliseconds. /// The maximum number of threads for the queue to use. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. protected ProcessQueue(ProcessItemFunctionSignature processItemFunction, CanProcessItemFunctionSignature canProcessItemFunction, double processInterval, int maximumThreads, int processTimeout, bool requeueOnTimeout, bool requeueOnException) : this(processItemFunction, null, canProcessItemFunction, new ConcurrentQueue(), processInterval, maximumThreads, processTimeout, requeueOnTimeout, requeueOnException) { } /// /// Creates a bulk item based on the generic List(Of T) class. /// /// Delegate that defines a method to process multiple items at once. /// Delegate that determines if an item can currently be processed. /// a value which represents the process interval in milliseconds. /// The maximum number of threads for the queue to use. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. protected ProcessQueue(ProcessItemsFunctionSignature processItemsFunction, CanProcessItemFunctionSignature canProcessItemFunction, double processInterval, int maximumThreads, int processTimeout, bool requeueOnTimeout, bool requeueOnException) : this(null, processItemsFunction, canProcessItemFunction, new ConcurrentQueue(), processInterval, maximumThreads, processTimeout, requeueOnTimeout, requeueOnException) { } /// /// Allows derived classes to define their own instance, if desired. /// /// Delegate that defines a method to process one item at a time. /// Delegate that defines a method to process multiple items at once. /// Delegate that determines if an item can currently be processed. /// A storage list for items to be processed. /// a value which represents the process interval in milliseconds. /// The maximum number of threads for the queue to use. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. protected ProcessQueue(ProcessItemFunctionSignature processItemFunction, ProcessItemsFunctionSignature processItemsFunction, CanProcessItemFunctionSignature canProcessItemFunction, IEnumerable processQueue, double processInterval, int maximumThreads, int processTimeout, bool requeueOnTimeout, bool requeueOnException) { m_processItemFunction = processItemFunction; // Defining this function creates a ProcessingStyle = OneAtATime process queue m_processItemsFunction = processItemsFunction; // Defining this function creates a ProcessingStyle = ManyAtOnce process queue m_canProcessItemFunction = canProcessItemFunction; m_processQueue = processQueue; m_maximumThreads = maximumThreads; m_processTimeout = processTimeout; m_requeueOnTimeout = requeueOnTimeout; m_requeueOnException = requeueOnException; if (processInterval == RealTimeProcessInterval) { // Instantiates process queue for real-time item processing m_processingIsRealTime = true; m_maximumThreads = 1; } else { // Instantiates process queue for intervaled item processing m_processTimer = new System.Timers.Timer(); m_processTimer.Elapsed += ProcessTimerThreadProc; m_processTimer.Interval = processInterval; m_processTimer.AutoReset = true; m_processTimer.Enabled = false; } } /// /// Releases the unmanaged resources before the object is reclaimed by . /// ~ProcessQueue() { Dispose(false); } #endregion #region [ Properties ] /// /// Gets or sets the user function for processing individual items in the one at a time. /// /// /// Cannot be defined simultaneously with . /// A must be defined to process a single item at a time or many items at once. /// Implementation makes ProcessingStyle = OneAtATime. /// public virtual ProcessItemFunctionSignature ProcessItemFunction { get { return m_processItemFunction; } set { if ((object)value != null) { m_processItemFunction = value; m_processItemsFunction = null; } } } /// /// Gets or sets the user function for processing multiple items in the at once. /// /// /// This function and cannot be defined at the same time /// A must be defined to process a single item at a time or many items at once /// Implementation of this function makes ProcessingStyle = ManyAtOnce /// public virtual ProcessItemsFunctionSignature ProcessItemsFunction { get { return m_processItemsFunction; } set { if ((object)value != null) { m_processItemsFunction = value; m_processItemFunction = null; } } } /// /// Gets or sets the user function determining if an item is ready to be processed. /// public virtual CanProcessItemFunctionSignature CanProcessItemFunction { get { return m_canProcessItemFunction; } set { m_canProcessItemFunction = value; } } /// /// Gets indicator that items will be processed in real-time. /// public virtual bool ProcessingIsRealTime { get { return m_processingIsRealTime; } } /// /// Gets the current for the (i.e., synchronous or asynchronous). /// /// /// The maximum number of processing threads determines the . /// If the maximum threads are set to one, item processing will be synchronous /// (i.e., ThreadingMode = Synchronous). /// If the maximum threads are more than one, item processing will be asynchronous /// (i.e., ThreadingMode = Asynchronous). /// /// Note that for asynchronous , the processing interval will control how many threads are spawned /// at once. If items are processed faster than the specified processing interval, only one process thread /// will ever be spawned at a time. To ensure multiple threads are utilized to items, lower /// the process interval (minimum process interval is 1 millisecond). /// /// public virtual QueueThreadingMode ThreadingMode { get { if (m_maximumThreads > 1) return QueueThreadingMode.Asynchronous; return QueueThreadingMode.Synchronous; } } /// /// Gets the item for the (i.e., one at a time or many at once). /// /// /// OneAtATime, if the is implemented. /// ManyAtOnce, if the is implemented. /// /// /// The implemented item processing function determines the . /// /// If the is ManyAtOnce, all available items in the are presented for processing /// at each processing interval. If you expect items to be processed in the order in which they were received, make /// sure you use a synchronous . Real-time are inherently synchronous. /// /// public virtual QueueProcessingStyle ProcessingStyle { get { if ((object)m_processItemFunction == null) return QueueProcessingStyle.ManyAtOnce; return QueueProcessingStyle.OneAtATime; } } /// /// Gets or sets the interval, in milliseconds, on which new items begin processing. /// public virtual double ProcessInterval { get { if (m_processingIsRealTime) return RealTimeProcessInterval; return m_processTimer.Interval; } set { if (m_processingIsRealTime) throw new InvalidOperationException("Cannot change process interval when " + Name + " is configured for real-time processing"); m_processTimer.Interval = value; } } /// /// Gets or sets the maximum number of threads to process simultaneously. /// /// Sets the maximum number of processing threads. /// Maximum number of processing threads. /// If MaximumThreads is set to one, item processing will be synchronous (i.e., ThreadingMode = Synchronous) public virtual int MaximumThreads { get { return m_maximumThreads; } set { if (m_processingIsRealTime) throw new InvalidOperationException("Cannot change the maximum number of threads when " + Name + " is configured for real-time processing"); m_maximumThreads = value; } } /// /// Gets or sets the maximum time, in milliseconds, allowed for processing an item. /// /// Sets the maximum number of milliseconds allowed to process an item. /// The maximum number of milliseconds allowed to process an item. /// Set to Timeout.Infinite (i.e., -1) to allow processing to take as long as needed. public virtual int ProcessTimeout { get { return m_processTimeout; } set { m_processTimeout = value; } } /// /// Gets or sets whether or not to automatically place an item back into the if the processing times out. /// /// Ignored if the ProcessTimeout is set to Timeout.Infinite (i.e., -1). public virtual bool RequeueOnTimeout { get { return m_requeueOnTimeout; } set { m_requeueOnTimeout = value; } } /// /// Gets or sets whether or not to automatically place an item back into the if an exception occurs /// while processing. /// public virtual bool RequeueOnException { get { return m_requeueOnException; } set { m_requeueOnException = value; } } /// /// Gets or sets indicator that the is currently enabled. /// public virtual bool Enabled { get { return m_enabled; } set { if (m_enabled && !value) Stop(); else if (!m_enabled && value) Start(); } } /// /// Gets indicator that the is actively processing items. /// public bool IsProcessing { get { if (m_processingIsRealTime) return ((object)m_realTimeProcessThread != null); lock (m_processTimer) { // Enabled flag changes are always in a critical section to ensure all items will be processed return m_processTimer.Enabled; } } } /// /// Gets the total number of items currently being processed. /// /// Total number of items currently being processed. public long ItemsBeingProcessed { get { return m_itemsProcessing; } } /// /// Gets the total number of items processed so far. /// /// Total number of items processed so far. public long TotalProcessedItems { get { return m_itemsProcessed; } } /// /// Gets the current number of active threads. /// /// Current number of active threads. public int ThreadCount { get { return m_threadCount; } } /// /// Gets the total amount of time, in seconds, that the has been active. /// public virtual Time RunTime { get { Ticks processingTime = 0; if (m_startTime > 0) { if (m_stopTime > 0) processingTime = m_stopTime - m_startTime; else processingTime = DateTime.Now.Ticks - m_startTime; } if (processingTime < 0) processingTime = 0; return processingTime.ToSeconds(); } } /// /// Gets or sets name for this . /// /// /// This name is used for class identification in strings (e.g., used in error messages). /// public virtual string Name { get { if (string.IsNullOrEmpty(m_name)) m_name = this.GetType().Name; return m_name; } set { m_name = value; } } /// Gets the number of elements actually contained in the . /// The number of elements actually contained in the . public virtual int Count { get { return m_processQueue.Count(); } } /// Gets a value indicating whether access to the is synchronized (thread safe). Always returns true for . /// true, is always synchronized (thread safe). /// The is effectively "synchronized" since all functions SyncLock operations internally. public bool IsSynchronized { get { return true; } } /// /// Gets the current run-time statistics of the as a single group of values. /// public virtual ProcessQueueStatistics CurrentStatistics { get { ProcessQueueStatistics statistics; statistics.IsEnabled = m_enabled; statistics.IsProcessing = IsProcessing; statistics.ProcessingInterval = ProcessInterval; statistics.ProcessingStyle = ProcessingStyle; statistics.ProcessTimeout = m_processTimeout; statistics.ThreadingMode = ThreadingMode; statistics.ActiveThreads = m_threadCount; statistics.ItemsBeingProcessed = m_itemsProcessing; statistics.TotalProcessedItems = m_itemsProcessed; statistics.QueueCount = Count; statistics.RunTime = RunTime; return statistics; } } /// /// Gets current status of . /// public virtual string Status { get { StringBuilder status = new StringBuilder(); status.Append(" Queue processing is: "); status.Append(m_enabled ? "Enabled" : "Disabled"); status.AppendLine(); status.Append(" Current processing state: "); status.Append(IsProcessing ? "Executing" : "Idle"); status.AppendLine(); status.Append(" Processing interval: "); if (m_processingIsRealTime) { status.Append("Real-time"); } else { status.Append(ProcessInterval); status.Append(" milliseconds"); } status.AppendLine(); status.Append(" Processing timeout: "); if (m_processTimeout == Timeout.Infinite) { status.Append("Infinite"); } else { status.Append(m_processTimeout); status.Append(" milliseconds"); } status.AppendLine(); status.Append(" Queue threading mode: "); if (ThreadingMode == QueueThreadingMode.Asynchronous) { status.Append("Asynchronous - "); status.Append(m_maximumThreads); status.Append(" maximum threads"); } else { status.Append("Synchronous"); } status.AppendLine(); status.Append(" Queue processing style: "); status.Append(ProcessingStyle == QueueProcessingStyle.OneAtATime ? "One at a time" : "Many at once"); status.AppendLine(); status.Append(" Total process run time: "); status.Append(RunTime.ToString()); status.AppendLine(); status.Append(" Total active threads: "); status.Append(m_threadCount); status.AppendLine(); status.Append(" Queued items to process: "); status.Append(Count); status.AppendLine(); status.Append(" Items being processed: "); status.Append(m_itemsProcessing); status.AppendLine(); status.Append(" Total items processed: "); status.Append(m_itemsProcessed); status.AppendLine(); return status.ToString(); } } /// /// Allows derived classes to access the interfaced internal directly. /// protected IEnumerable InternalQueue { get { return m_processQueue; } } /// /// Gets an object that can be used to synchronize access to the . /// public object SyncRoot { get { return m_processQueue; } } #endregion #region [ Methods ] /// /// 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 { // Must stop thread, otherwise your app will keep running :) Stop(); if (disposing) { if ((object)m_processTimer != null) { m_processTimer.Elapsed -= ProcessTimerThreadProc; m_processTimer.Dispose(); } m_processTimer = null; m_processQueue = null; m_processItemFunction = null; m_processItemsFunction = null; m_canProcessItemFunction = null; } } finally { m_disposed = true; // Prevent duplicate dispose. if (Disposed != null) Disposed(this, EventArgs.Empty); } } } /// /// Adds an item to the . /// /// The item to add to the . public virtual void Add(T item) { ConcurrentQueue queue = m_processQueue as ConcurrentQueue; if ((object)queue != null) { queue.Enqueue(item); if (m_enabled && m_processingIsRealTime && Interlocked.CompareExchange(ref m_processing, 1, 0) == 0) { if (m_processQueue.Any()) ThreadPool.QueueUserWorkItem(RealTimeDataProcessingLoop); else Interlocked.Exchange(ref m_processing, 0); } DataAdded(); } else { throw new InvalidOperationException(); } } /// /// Adds the specified to the . /// /// The elements to be added to the . public virtual void AddRange(IEnumerable items) { foreach (T item in items) { Add(item); } } /// /// Determines whether an element is in the . /// /// True, if item is found in the ; otherwise, false. /// The object to locate in the . The value can be null for reference types. public virtual bool Contains(T item) { return m_processQueue.Contains(item); } /// /// Copies the elements of the to an , starting at a particular index. /// /// /// The one-dimensional that is the destination of the elements copied from the . /// The array must have zero-based indexing. /// /// The zero-based index in array at which copying begins. public virtual void CopyTo(T[] array, int index) { m_processQueue.ToList().CopyTo(array, index); } /// /// Copies the elements contained in the to a new array. /// /// A new array containing the elements copied from the . public virtual T[] ToArray() { return m_processQueue.ToArray(); } /// /// Attempts to add an object to the . /// /// The object to add to the . /// true if the object was successfully added; otherwise, false. public virtual bool TryAdd(T item) { Add(item); return true; } /// /// Attempts to remove and return an object from the . /// /// When this method returns, if the object was removed and returned successfully, item contains the removed object. If no object was available to be removed, the value is unspecified. /// true if an object was removed and returned successfully; otherwise, false. public virtual bool TryTake(out T item) { ConcurrentQueue queue = m_processQueue as ConcurrentQueue; if ((object)queue != null) return queue.TryDequeue(out item); throw new InvalidOperationException(); } /// /// Attempts to remove and return all objects from the . /// /// When this method returns, if any objects were removed and returned successfully, item array contains the removed objects. If no object was available to be removed, the value is null. /// true if any objects were removed and returned successfully; otherwise, false. public virtual bool TryTake(out T[] items) { ConcurrentQueue queue = m_processQueue as ConcurrentQueue; if ((object)queue != null) { T item; List taken = new List(); while (queue.TryDequeue(out item)) { taken.Add(item); } if (taken.Count > 0) { items = taken.ToArray(); return true; } items = null; return false; } throw new InvalidOperationException(); } /// /// Returns an enumerator that iterates through the . /// /// An enumerator for the . public virtual IEnumerator GetEnumerator() { return m_processQueue.GetEnumerator(); } /// /// Removes all elements from the . /// public virtual void Clear() { ConcurrentQueue queue = m_processQueue as ConcurrentQueue; if ((object)queue != null) { T result; while (queue.TryDequeue(out result)) { } } else { throw new InvalidOperationException(); } } /// /// Starts item processing. /// public virtual void Start() { m_enabled = true; m_threadCount = 0; m_itemsProcessed = 0; m_stopTime = 0; m_startTime = DateTime.Now.Ticks; // Note that real-time queues have their main thread running continually, but for // intervaled queues, processing occurs only when data is available to be processed. if (m_processingIsRealTime) { // Start real-time processing thread #if ThreadTracking m_realTimeProcessThread = new ManagedThread(RealTimeThreadProc); m_realTimeProcessThread.Name = "TVA.Collections.ProcessQueue.RealTimeThreadProc() [" + Name + "]"; #else m_realTimeProcessThread = new Thread(RealTimeThreadProc); #endif m_realTimeProcessThread.IsBackground = true; m_realTimeProcessThread.Start(); } else { // Start intervaled processing, if there items in the queue lock (m_processTimer) { // Enabled flag changes are always in a critical section to ensure all items will be processed m_processTimer.Enabled = Count > 0; } } } void ISupportLifecycle.Initialize() { // Enabled property handles check for redundant calls... Enabled = true; } /// /// Stops item processing. /// public virtual void Stop() { m_enabled = false; if (m_processingIsRealTime) { // Remove reference to process thread - it will stop gracefully after it has finished processing // current set of items since enabled is false... m_realTimeProcessThread = null; } else { // Stops intervaled processing, if active. if ((object)m_processTimer != null) { lock (m_processTimer) { // Enabled flag changes are always in a critical section to ensure all items will be processed m_processTimer.Enabled = false; } } } m_stopTime = DateTime.Now.Ticks; } /// /// Blocks the current thread, if the is active (i.e., user has called "Start" method), until all items /// in are processed, and then stops the . /// /// /// /// Begins processing items as quickly as possible, regardless of currently defined process interval, until all /// items in the have been processed. Stops the when this function ends. /// This method is typically called on shutdown to make sure any remaining queued items get processed before the /// is destructed. /// /// /// It is possible for items to be added to the while the flush is executing. The flush will continue to /// process items as quickly as possible until the is empty. Unless the user stops queueing items to be /// processed, the flush call may never return (not a happy situtation on shutdown). For this reason, during this /// function call, requeueing of items on exception or process timeout is temporarily disabled. /// /// /// The does not clear queue prior to destruction. If the user fails to call this method before the /// class is destructed, there may be items that remain unprocessed in the . /// /// public virtual void Flush() { bool enabled = m_enabled; // Stop all queue processing... Stop(); if (enabled) { bool originalRequeueOnTimeout = m_requeueOnTimeout; bool originalRequeueOnException = m_requeueOnException; // We must disable requeueing of items or this method could continue indefinitely. m_requeueOnTimeout = false; m_requeueOnException = false; // Only waits around if there is something to process. while (Count > 0) { // Create a real-time processing loop that will process remaining items as quickly as possible. while (m_processQueue.Any()) { if ((object)m_processItemsFunction == null) { // Processes one item at a time. ProcessNextItem(); } else { // Processes multiple items at once. ProcessNextItems(); } } } // Just in case user continues to use queue after flush, this restores original states. m_requeueOnTimeout = originalRequeueOnTimeout; m_requeueOnException = originalRequeueOnException; } } IEnumerator IEnumerable.GetEnumerator() { return ((IEnumerable)m_processQueue).GetEnumerator(); } void ICollection.CopyTo(Array array, int index) { m_processQueue.ToArray().CopyTo(array, index); } #region [ Item Processing Functions ] /// /// Raises the base class event. /// /// /// Derived classes cannot raise events of their base classes, so we expose event wrapper methods to accomodate /// as needed. /// /// A generic type T to be passed to ItemProcessed. protected virtual void OnItemProcessed(T item) { if ((object)ItemProcessed != null) ItemProcessed(this, new EventArgs(item)); } /// /// Raises the base class event. /// /// /// Derived classes cannot raise events of their base classes, so we expose event wrapper methods to accomodate /// as needed. /// /// An array of generic type T to be passed to ItemsProcessed. protected virtual void OnItemsProcessed(T[] items) { if ((object)ItemsProcessed != null) ItemsProcessed(this, new EventArgs(items)); } /// /// Raises the base class event. /// /// /// Derived classes cannot raise events of their base classes, so we expose event wrapper methods to accomodate /// as needed. /// /// A generic type T to be passed to ItemProcessed. protected virtual void OnItemTimedOut(T item) { if ((object)ItemTimedOut != null) ItemTimedOut(this, new EventArgs(item)); } /// /// Raises the base class event. /// /// /// Derived classes cannot raise events of their base classes, so we expose event wrapper methods to accomodate /// as needed. /// /// An array of generic type T to be passed to ItemsProcessed. protected virtual void OnItemsTimedOut(T[] items) { if ((object)ItemsTimedOut != null) ItemsTimedOut(this, new EventArgs(items)); } /// /// Raises the base class event. /// /// /// Derived classes cannot raise events of their base classes, so we expose event wrapper methods to accomodate /// as needed. /// /// to be passed to ProcessException. protected virtual void OnProcessException(Exception ex) { if ((object)ProcessException != null) ProcessException(this, new EventArgs(ex)); } /// /// Notifies a class that data was added, so it can begin processing data. /// /// /// /// Derived classes *must* make sure to call this method after data gets added, so that the /// process timer can be enabled for intervaled and data processing can begin. /// /// /// To make sure items in the always get processed, this function is expected to be /// invoked from within a SyncLock of the exposed SyncRoot (i.e., InternalList). /// /// protected virtual void DataAdded() { // For queues that are not processing in real-time, we start the intervaled process timer // when data is added, if it's not running already if (!m_processingIsRealTime) { lock (m_processTimer) { // Enabled flag changes are always in a critical section to ensure all items will be processed if (m_enabled && !m_processTimer.Enabled) m_processTimer.Enabled = true; } } } /// /// Determines if an item can be processed. /// /// True, if user provided no implementation for the CanProcessItemFunction. /// /// /// Use this function instead of invoking the CanProcessItemFunction pointer /// directly, since implementation of this delegate is optional. /// /// /// The item T to process. /// A value indicating whether it can process the item or not. protected virtual bool CanProcessItem(T item) { // If user provided no implementation for this function or function failed, we assume item can be processed. if ((object)m_canProcessItemFunction == null) return true; try { // When user function is provided, we call it to determine if item state allows processing at this time. return m_canProcessItemFunction(item); } catch (ThreadAbortException) { // Rethrow thread abort so calling method can respond appropriately throw; } catch (Exception ex) { // Processing will not stop for any errors thrown by the user function, but errors will be reported. OnProcessException(ex); } // Assuming processing must go on if the user function fails return true; } /// /// Determines if all items can be processed. /// /// True, if user provided no implementation for the CanProcessItemFunction. /// /// /// Use this function instead of invoking the CanProcessItemFunction pointer /// directly, since implementation of this delegate is optional. /// /// /// An array of items of type T. /// A value indicating whether the process queue can process the items. protected virtual bool CanProcessItems(T[] items) { // If user provided no implementation for this function or function failed, we assume item can be processed. if ((object)m_canProcessItemFunction == null) return true; // Otherwise we call user function for each item to determine if all items are ready for processing. bool allItemsCanBeProcessed = true; foreach (T item in items) { if (!CanProcessItem(item)) { allItemsCanBeProcessed = false; break; } } return allItemsCanBeProcessed; } /// /// Requeues item into according to specified requeue reason. /// /// A generic item of type T to be requeued. /// The reason the object is being requeued. protected virtual void RequeueItem(T item, RequeueReason reason) { Add(item); } /// /// Requeues items into according to specified requeue reason. /// /// Array of type T to be requeued. /// The reason the object is being requeued. protected virtual void RequeueItems(T[] items, RequeueReason reason) { foreach (T item in items) { Add(item); } } /// /// Handles standard processing of a single item. /// /// A generic item of type T to be processed. private void ProcessItem(T item) { try { // Invokes user function to process item. m_processItemFunction(item); Interlocked.Increment(ref m_itemsProcessed); // Notifies consumers of successfully processed items. OnItemProcessed(item); } catch (ThreadAbortException) { // Rethrows thread abort, so calling method can respond appropriately. throw; } catch (Exception ex) { // Requeues item on processing exception, if requested. if (m_requeueOnException) RequeueItem(item, RequeueReason.Exception); // Processing will not stop for any errors thrown by the user function, but errors will be reported. OnProcessException(ex); } } /// /// Handles standard processing of multiple items. /// /// Array of type T. private void ProcessItems(T[] items) { try { // Invokes user function to process items. m_processItemsFunction(items); Interlocked.Add(ref m_itemsProcessed, items.Length); // Notifies consumers of successfully processed items. OnItemsProcessed(items); } catch (ThreadAbortException) { // Rethrows thread abort, so calling method can respond appropriately. throw; } catch (Exception ex) { // Requeues items on processing exception, if requested. if (m_requeueOnException) RequeueItems(items, RequeueReason.Exception); // Processing will not stop for any errors thrown by the user function, but errors will be reported. OnProcessException(ex); } } /// /// Creates a real-time thread for processing items. /// private void RealTimeThreadProc() { int processing; int sleepTime = 1; long noWorkSleeps = 0L; // Creates a real-time processing loop that will start item processing as quickly as possible. while (m_enabled) { processing = Interlocked.CompareExchange(ref m_processing, 1, 0); // Kick start processing when items exist that are not currently being processed if (processing == 0 && m_processQueue.Any()) { sleepTime = 1; noWorkSleeps = 0L; ThreadPool.QueueUserWorkItem(RealTimeDataProcessingLoop); } else { // If the processing flag was set but no items were found in the process queue, // the asynchronous loop was never spawned so we need to clear the processing flag if (processing == 0) Interlocked.Exchange(ref m_processing, 0); // Vary sleep time based on how often kick start is being processed, up to one second for very idle queues if (noWorkSleeps > 1000L) sleepTime = 1000; // It will take well over 1.5 minutes of no work before sleeping for 1 second else if (noWorkSleeps > 100L) sleepTime = 100; // It will take at least one second of no work before sleeping for 100ms else if (noWorkSleeps > 5L) sleepTime = 10; // It will take at least 5ms of no work before sleeping for 10ms noWorkSleeps++; // Wait around for more items to process Thread.Sleep(sleepTime); } } } // Creates a real-time loop for processing data that runs as long as there is data to process private void RealTimeDataProcessingLoop(object state) { if ((object)m_processItemsFunction == null) ProcessNextItem(); else ProcessNextItems(); if (m_enabled && m_processQueue.Any()) ThreadPool.QueueUserWorkItem(RealTimeDataProcessingLoop); else Interlocked.Exchange(ref m_processing, 0); } /// /// Processes queued items on an interval. /// /// The sender object of the item. /// Arguments for the elapsed event. private void ProcessTimerThreadProc(object sender, System.Timers.ElapsedEventArgs e) { // The system timer creates an intervaled processing loop such that if an existing item processing // call hasn't completed before next interval, multiple processing calls will be spawned thereby // distributing item processing across multiple threads as needed. if ((object)m_processItemsFunction == null) { // Process one item at a time. ProcessNextItem(); } else { // Process multiple items at once. ProcessNextItems(); } if ((object)m_processTimer != null) { // Stop the process timer if there is no more data to process. lock (m_processTimer) { // Enabled flag changes are always in a critical section to ensure all items will be processed if (!m_processQueue.Any()) m_processTimer.Enabled = false; } } } /// /// Processes next item in queue, one at a time (i.e., ProcessingStyle = OneAtATime). /// private void ProcessNextItem() { T nextItem = default(T); bool processingItem = false; try { // Get the next item to be processed if the number of current process threads is less // than the maximum allowable number of process threads. if (m_threadCount < m_maximumThreads && TryTake(out nextItem)) { // Call optional user function to see if we can process this item. if (CanProcessItem(nextItem)) { Interlocked.Increment(ref m_threadCount); Interlocked.Increment(ref m_itemsProcessing); processingItem = true; } else { // If item state is not ready for processing, all we can do is requeue. RequeueItem(nextItem, RequeueReason.CannotProcess); } } if (processingItem) { if (m_processTimeout == Timeout.Infinite) { // If an item is in the queue to process, and the process queue was not set up with a process // timeout, we use the current thread (i.e., the timer event or real-time thread) to process the // next item taking as long as we need for it to complete. For timer events, the next item in // the queue will begin processing even if this item is not completed, but no more than the // specified number of maximum threads will ever be spawned at once. ProcessItem(nextItem); } else { // If an item is in the queue to process, with a specified process timeout, a new thread is // created to handle the processing. The timer event or real-time thread that invoked this method // is already a new thread, so the only reason to create another thread is to implement the // process timeout if the process takes too long to run. This is done by joining the current // thread (which will block it) until the specified interval has passed or the process thread // completes, whichever comes first. This is a safe operation since the current thread // (i.e., the timer event or real-time thread) was already an independent thread and will not // block any other processing, including another timer event. if (!TemporalTask.Process(this, nextItem)) { // Notify user of process timeout, in case they want to do anything special. OnItemTimedOut(nextItem); // Requeues item on processing timeout, if requested. if (m_requeueOnTimeout) RequeueItem(nextItem, RequeueReason.Timeout); } } } } catch (ThreadAbortException) { // Rethrows thread abort, so calling method can respond appropriately. throw; } catch (Exception ex) { // Processing will not stop for any errors encountered here, but errors will be reported. OnProcessException(ex); } finally { // Decrements thread count, if item was retrieved for processing. if (processingItem) { Interlocked.Decrement(ref m_threadCount); Interlocked.Decrement(ref m_itemsProcessing); } } } /// /// Processes next items in an array of items as a group (i.e., ProcessingStyle = ManyAtOnce). /// private void ProcessNextItems() { T[] nextItems = null; bool processingItems = false; try { // Get next items to be processed if the number of current process threads is less // than the maximum allowable number of process threads. if (m_threadCount < m_maximumThreads && TryTake(out nextItems)) { // Call optional user function to see if these items can be processed. if (CanProcessItems(nextItems)) { Interlocked.Increment(ref m_threadCount); Interlocked.Add(ref m_itemsProcessing, nextItems.Length); processingItems = true; } else { // If item state is not ready for processing, all we can do is requeue. RequeueItems(nextItems, RequeueReason.CannotProcess); } } if (processingItems) { if (m_processTimeout == Timeout.Infinite) { // If items are in the queue to process, and the process queue was not set up with a process // timeout, the current thread (i.e., the timer event or real-time thread) is used to process the // next items taking as long as necessary to complete. For timer events, any new items available // in the queue will be processed, even if the current items have not completed, but no more than // the specified number of maximum threads will ever be spawned at once. ProcessItems(nextItems); } else { // If items are in the queue to process, and a process timeout was specified, a new thread is // created to handle the processing. The timer event or real-time thread that invoked this method // is already a new thread, so the only reason to create another thread is to implement the // process timeout if the process takes too long to run. We do this by joining the current thread // (which will block it) until the specified interval has passed or the process thread completes, // whichever comes first. This is a safe operation, since the current thread (i.e., the timer // event or real-time thread) was already an independent thread and will not block any other // processing, including another timer event. if (!TemporalTask.Process(this, nextItems)) { // Notify the user of the process timeout, in case they want to do anything special. OnItemsTimedOut(nextItems); // Requeues items on processing timeout, if requested. if (m_requeueOnTimeout) RequeueItems(nextItems, RequeueReason.Timeout); } } } } catch (ThreadAbortException) { // Rethrows thread abort, so calling method can respond appropriately. throw; } catch (Exception ex) { // Processing will not stop for any errors encountered here, but errors will be reported. OnProcessException(ex); } finally { // Decrements thread count, if items were retrieved for processing. if (processingItems) { Interlocked.Decrement(ref m_threadCount); Interlocked.Add(ref m_itemsProcessing, -nextItems.Length); } } } #endregion #endregion #region [ Static ] #region [ Single-Item Processing Constructors ] /// /// Creates a new asynchronous with the default settings: ProcessInterval = 100, MaximumThreads = 5, /// ProcessTimeout = Infinite, RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that processes one item at a time. /// A ProcessQueue object based on type T. public static ProcessQueue CreateAsynchronousQueue(ProcessItemFunctionSignature processItemFunction) { return CreateAsynchronousQueue(processItemFunction, null, DefaultProcessInterval, DefaultMaximumThreads, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new asynchronous with the default settings: ProcessInterval = 100, MaximumThreads = 5, /// ProcessTimeout = Infinite, RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that processes one item at a time. /// Delegate which determines whether an item can be processed. /// A ProcessQueue object based on type T. public static ProcessQueue CreateAsynchronousQueue(ProcessItemFunctionSignature processItemFunction, CanProcessItemFunctionSignature canProcessItemFunction) { return CreateAsynchronousQueue(processItemFunction, canProcessItemFunction, DefaultProcessInterval, DefaultMaximumThreads, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new asynchronous with the default settings: ProcessInterval = 100, /// ProcessTimeout = Infinite, RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that processes one item at a time. /// An value that determines the maximum number of threads used to process items. /// A ProcessQueue object based on type T. public static ProcessQueue CreateAsynchronousQueue(ProcessItemFunctionSignature processItemFunction, int maximumThreads) { return CreateAsynchronousQueue(processItemFunction, null, DefaultProcessInterval, maximumThreads, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new asynchronous with the default settings: ProcessInterval = 100, /// ProcessTimeout = Infinite, RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that processes one item at a time. /// Delegate which determines whether an item can be processed. /// An value that determines the maximum number of threads used to process items. /// A ProcessQueue object based on type T. public static ProcessQueue CreateAsynchronousQueue(ProcessItemFunctionSignature processItemFunction, CanProcessItemFunctionSignature canProcessItemFunction, int maximumThreads) { return CreateAsynchronousQueue(processItemFunction, canProcessItemFunction, DefaultProcessInterval, maximumThreads, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new asynchronous using specified settings. /// /// Delegate that processes one item at a time. /// a value which represents the process interval in milliseconds. /// An value that determines the maximum number of threads used to process items. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. /// A ProcessQueue object based on type T. public static ProcessQueue CreateAsynchronousQueue(ProcessItemFunctionSignature processItemFunction, double processInterval, int maximumThreads, int processTimeout, bool requeueOnTimeout, bool requeueOnException) { return CreateAsynchronousQueue(processItemFunction, null, processInterval, maximumThreads, processTimeout, requeueOnTimeout, requeueOnException); } /// /// Creates a new asynchronous using specified settings. /// /// Delegate that processes one item at a time. /// Delegate which determines whether an item can be processed. /// a value which represents the process interval in milliseconds. /// An value that determines the maximum number of threads used to process items. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. /// A ProcessQueue object based on type T. public static ProcessQueue CreateAsynchronousQueue(ProcessItemFunctionSignature processItemFunction, CanProcessItemFunctionSignature canProcessItemFunction, double processInterval, int maximumThreads, int processTimeout, bool requeueOnTimeout, bool requeueOnException) { return new ProcessQueue(processItemFunction, canProcessItemFunction, processInterval, maximumThreads, processTimeout, requeueOnTimeout, requeueOnException); } /// /// Creates a new synchronous (i.e., single process thread) with the default settings: /// ProcessInterval = 100, ProcessTimeout = Infinite, RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that processes one item at a time. /// A ProcessQueue object based on type T. public static ProcessQueue CreateSynchronousQueue(ProcessItemFunctionSignature processItemFunction) { return CreateSynchronousQueue(processItemFunction, null, DefaultProcessInterval, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new synchronous (i.e., single process thread) with the default settings: /// ProcessInterval = 100, ProcessTimeout = Infinite, RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that processes one item at a time. /// Delegate which determines whether an item can be processed. /// A ProcessQueue object based on type T. public static ProcessQueue CreateSynchronousQueue(ProcessItemFunctionSignature processItemFunction, CanProcessItemFunctionSignature canProcessItemFunction) { return CreateSynchronousQueue(processItemFunction, canProcessItemFunction, DefaultProcessInterval, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new synchronous (i.e., single process thread) using specified settings. /// /// Delegate that processes one item at a time. /// a value which represents the process interval in milliseconds. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. /// A ProcessQueue object based on type T. public static ProcessQueue CreateSynchronousQueue(ProcessItemFunctionSignature processItemFunction, double processInterval, int processTimeout, bool requeueOnTimeout, bool requeueOnException) { return CreateSynchronousQueue(processItemFunction, null, processInterval, processTimeout, requeueOnTimeout, requeueOnException); } /// /// Creates a new synchronous (i.e., single process thread) using specified settings. /// /// Delegate that processes one item at a time. /// Delegate which determines whether an item can be processed. /// a value which represents the process interval in milliseconds. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. /// A ProcessQueue object based on type T. public static ProcessQueue CreateSynchronousQueue(ProcessItemFunctionSignature processItemFunction, CanProcessItemFunctionSignature canProcessItemFunction, double processInterval, int processTimeout, bool requeueOnTimeout, bool requeueOnException) { return new ProcessQueue(processItemFunction, canProcessItemFunction, processInterval, 1, processTimeout, requeueOnTimeout, requeueOnException); } /// /// Creates a new real-time with the default settings: ProcessTimeout = Infinite, /// RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that processes one item at a time. /// A ProcessQueue object based on type T. public static ProcessQueue CreateRealTimeQueue(ProcessItemFunctionSignature processItemFunction) { return CreateRealTimeQueue(processItemFunction, null, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new real-time with the default settings: ProcessTimeout = Infinite, /// RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that processes one item at a time. /// Delegate which determines whether an item can be processed. /// A ProcessQueue object based on type T. public static ProcessQueue CreateRealTimeQueue(ProcessItemFunctionSignature processItemFunction, CanProcessItemFunctionSignature canProcessItemFunction) { return CreateRealTimeQueue(processItemFunction, canProcessItemFunction, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new real-time using specified settings. /// /// Delegate that processes one item at a time. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. /// A ProcessQueue object based on type T. public static ProcessQueue CreateRealTimeQueue(ProcessItemFunctionSignature processItemFunction, int processTimeout, bool requeueOnTimeout, bool requeueOnException) { return CreateRealTimeQueue(processItemFunction, null, processTimeout, requeueOnTimeout, requeueOnException); } /// /// Creates a new real-time using specified settings. /// /// Delegate that processes one item at a time. /// Delegate which determines whether an item can be processed. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. /// A ProcessQueue object based on type T. public static ProcessQueue CreateRealTimeQueue(ProcessItemFunctionSignature processItemFunction, CanProcessItemFunctionSignature canProcessItemFunction, int processTimeout, bool requeueOnTimeout, bool requeueOnException) { return new ProcessQueue(processItemFunction, canProcessItemFunction, RealTimeProcessInterval, 1, processTimeout, requeueOnTimeout, requeueOnException); } #endregion #region [ Multi-Item Processing Constructors ] /// /// Creates a new asynchronous, bulk item with the default settings: ProcessInterval = 100, /// MaximumThreads = 5, ProcessTimeout = Infinite, RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that defines a method to process multiple items at once. /// A ProcessQueue object based on type T. public static ProcessQueue CreateAsynchronousQueue(ProcessItemsFunctionSignature processItemsFunction) { return CreateAsynchronousQueue(processItemsFunction, null, DefaultProcessInterval, DefaultMaximumThreads, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new asynchronous, bulk item with the default settings: ProcessInterval = 100, /// MaximumThreads = 5, ProcessTimeout = Infinite, RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that defines a method to process multiple items at once. /// Delegate which determines whether an item can be processed. /// A ProcessQueue object based on type T. public static ProcessQueue CreateAsynchronousQueue(ProcessItemsFunctionSignature processItemsFunction, CanProcessItemFunctionSignature canProcessItemFunction) { return CreateAsynchronousQueue(processItemsFunction, canProcessItemFunction, DefaultProcessInterval, DefaultMaximumThreads, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new asynchronous, bulk item with the default settings: ProcessInterval = 100, /// ProcessTimeout = Infinite, RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that defines a method to process multiple items at once. /// An value that determines the maximum number of threads used to process items. /// A ProcessQueue object based on type T. public static ProcessQueue CreateAsynchronousQueue(ProcessItemsFunctionSignature processItemsFunction, int maximumThreads) { return CreateAsynchronousQueue(processItemsFunction, null, DefaultProcessInterval, maximumThreads, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new asynchronous, bulk item with the default settings: ProcessInterval = 100, /// ProcessTimeout = Infinite, RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that defines a method to process multiple items at once. /// Delegate which determines whether an item can be processed. /// An value that determines the maximum number of threads used to process items. /// A ProcessQueue object based on type T. public static ProcessQueue CreateAsynchronousQueue(ProcessItemsFunctionSignature processItemsFunction, CanProcessItemFunctionSignature canProcessItemFunction, int maximumThreads) { return CreateAsynchronousQueue(processItemsFunction, canProcessItemFunction, DefaultProcessInterval, maximumThreads, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new asynchronous, bulk item using specified settings. /// /// Delegate that defines a method to process multiple items at once. /// a value which represents the process interval in milliseconds. /// An value that determines the maximum number of threads used to process items. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. /// A ProcessQueue object based on type T. public static ProcessQueue CreateAsynchronousQueue(ProcessItemsFunctionSignature processItemsFunction, double processInterval, int maximumThreads, int processTimeout, bool requeueOnTimeout, bool requeueOnException) { return CreateAsynchronousQueue(processItemsFunction, null, processInterval, maximumThreads, processTimeout, requeueOnTimeout, requeueOnException); } /// /// Creates a new asynchronous, bulk item using specified settings. /// /// Delegate that defines a method to process multiple items at once. /// Delegate which determines whether an item can be processed. /// a value which represents the process interval in milliseconds. /// An value that determines the maximum number of threads used to process items. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. /// A ProcessQueue object based on type T. public static ProcessQueue CreateAsynchronousQueue(ProcessItemsFunctionSignature processItemsFunction, CanProcessItemFunctionSignature canProcessItemFunction, double processInterval, int maximumThreads, int processTimeout, bool requeueOnTimeout, bool requeueOnException) { return new ProcessQueue(processItemsFunction, canProcessItemFunction, processInterval, maximumThreads, processTimeout, requeueOnTimeout, requeueOnException); } /// /// Creates a new synchronous, bulk item (i.e., single process thread) with the default settings: /// ProcessInterval = 100, ProcessTimeout = Infinite, RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that defines a method to process multiple items at once. /// A ProcessQueue object based on type T. public static ProcessQueue CreateSynchronousQueue(ProcessItemsFunctionSignature processItemsFunction) { return CreateSynchronousQueue(processItemsFunction, null, DefaultProcessInterval, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new synchronous, bulk item (i.e., single process thread) with the default settings: /// ProcessInterval = 100, ProcessTimeout = Infinite, RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that defines a method to process multiple items at once. /// Delegate which determines whether an item can be processed. /// A ProcessQueue object based on type T. public static ProcessQueue CreateSynchronousQueue(ProcessItemsFunctionSignature processItemsFunction, CanProcessItemFunctionSignature canProcessItemFunction) { return CreateSynchronousQueue(processItemsFunction, canProcessItemFunction, DefaultProcessInterval, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new synchronous, bulk item (i.e., single process thread) using specified settings. /// /// Delegate that defines a method to process multiple items at once. /// a value which represents the process interval in milliseconds. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. /// A ProcessQueue object based on type T. public static ProcessQueue CreateSynchronousQueue(ProcessItemsFunctionSignature processItemsFunction, double processInterval, int processTimeout, bool requeueOnTimeout, bool requeueOnException) { return CreateSynchronousQueue(processItemsFunction, null, processInterval, processTimeout, requeueOnTimeout, requeueOnException); } /// /// Creates a new synchronous, bulk item (i.e., single process thread) using specified settings. /// /// Delegate that defines a method to process multiple items at once. /// Delegate which determines whether an item can be processed. /// a value which represents the process interval in milliseconds. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. /// A ProcessQueue object based on type T. public static ProcessQueue CreateSynchronousQueue(ProcessItemsFunctionSignature processItemsFunction, CanProcessItemFunctionSignature canProcessItemFunction, double processInterval, int processTimeout, bool requeueOnTimeout, bool requeueOnException) { return new ProcessQueue(processItemsFunction, canProcessItemFunction, processInterval, 1, processTimeout, requeueOnTimeout, requeueOnException); } /// /// Creates a new real-time, bulk item with the default settings: ProcessTimeout = Infinite, /// RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that defines a method to process multiple items at once. /// A ProcessQueue object based on type T. public static ProcessQueue CreateRealTimeQueue(ProcessItemsFunctionSignature processItemsFunction) { return CreateRealTimeQueue(processItemsFunction, null, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new real-time, bulk item with the default settings: ProcessTimeout = Infinite, /// RequeueOnTimeout = False, RequeueOnException = False. /// /// Delegate that defines a method to process multiple items at once. /// Delegate which determines whether an item can be processed. /// A ProcessQueue object based on type T. public static ProcessQueue CreateRealTimeQueue(ProcessItemsFunctionSignature processItemsFunction, CanProcessItemFunctionSignature canProcessItemFunction) { return CreateRealTimeQueue(processItemsFunction, canProcessItemFunction, DefaultProcessTimeout, DefaultRequeueOnTimeout, DefaultRequeueOnException); } /// /// Creates a new real-time, bulk item using specified settings. /// /// Delegate that defines a method to process multiple items at once. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. /// A ProcessQueue object based on type T. public static ProcessQueue CreateRealTimeQueue(ProcessItemsFunctionSignature processItemsFunction, int processTimeout, bool requeueOnTimeout, bool requeueOnException) { return CreateRealTimeQueue(processItemsFunction, null, processTimeout, requeueOnTimeout, requeueOnException); } /// /// Creates a new real-time, bulk item using specified settings. /// /// Delegate that defines a method to process multiple items at once. /// Delegate which determines whether an item can be processed. /// The number of seconds before a process should timeout. /// A value that indicates whether a process should requeue an item on timeout. /// A value that indicates whether a process should requeue after an exception. /// A ProcessQueue object based on type T. public static ProcessQueue CreateRealTimeQueue(ProcessItemsFunctionSignature processItemsFunction, CanProcessItemFunctionSignature canProcessItemFunction, int processTimeout, bool requeueOnTimeout, bool requeueOnException) { return new ProcessQueue(processItemsFunction, canProcessItemFunction, RealTimeProcessInterval, 1, processTimeout, requeueOnTimeout, requeueOnException); } #endregion #endregion } }