123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737 |
- #region Release History
- // Smart Thread Pool
- // 7 Aug 2004 - Initial release
- //
- // 14 Sep 2004 - Bug fixes
- //
- // 15 Oct 2004 - Added new features
- // - Work items return result.
- // - Support waiting synchronization for multiple work items.
- // - Work items can be cancelled.
- // - Passage of the caller thread’s context to the thread in the pool.
- // - Minimal usage of WIN32 handles.
- // - Minor bug fixes.
- //
- // 26 Dec 2004 - Changes:
- // - Removed static constructors.
- // - Added finalizers.
- // - Changed Exceptions so they are serializable.
- // - Fixed the bug in one of the SmartThreadPool constructors.
- // - Changed the SmartThreadPool.WaitAll() so it will support any number of waiters.
- // The SmartThreadPool.WaitAny() is still limited by the .NET Framework.
- // - Added PostExecute with options on which cases to call it.
- // - Added option to dispose of the state objects.
- // - Added a WaitForIdle() method that waits until the work items queue is empty.
- // - Added an STPStartInfo class for the initialization of the thread pool.
- // - Changed exception handling so if a work item throws an exception it
- // is rethrown at GetResult(), rather then firing an UnhandledException event.
- // Note that PostExecute exception are always ignored.
- //
- // 25 Mar 2005 - Changes:
- // - Fixed lost of work items bug
- //
- // 3 Jul 2005: Changes.
- // - Fixed bug where Enqueue() throws an exception because PopWaiter() returned null, hardly reconstructed.
- //
- // 16 Aug 2005: Changes.
- // - Fixed bug where the InUseThreads becomes negative when canceling work items.
- //
- // 31 Jan 2006 - Changes:
- // - Added work items priority
- // - Removed support of chained delegates in callbacks and post executes (nobody really use this)
- // - Added work items groups
- // - Added work items groups idle event
- // - Changed SmartThreadPool.WaitAll() behavior so when it gets empty array
- // it returns true rather then throwing an exception.
- // - Added option to start the STP and the WIG as suspended
- // - Exception behavior changed, the real exception is returned by an
- // inner exception
- // - Added option to keep the Http context of the caller thread. (Thanks to Steven T.)
- // - Added performance counters
- // - Added priority to the threads in the pool
- //
- // 13 Feb 2006 - Changes:
- // - Added a call to the dispose of the Performance Counter so
- // their won't be a Performance Counter leak.
- // - Added exception catch in case the Performance Counters cannot
- // be created.
- //
- // 17 May 2008 - Changes:
- // - Changed the dispose behavior and removed the Finalizers.
- // - Enabled the change of the MaxThreads and MinThreads at run time.
- // - Enabled the change of the Concurrency of a IWorkItemsGroup at run
- // time If the IWorkItemsGroup is a SmartThreadPool then the Concurrency
- // refers to the MaxThreads.
- // - Improved the cancel behavior.
- // - Added events for thread creation and termination.
- // - Fixed the HttpContext context capture.
- // - Changed internal collections so they use generic collections
- // - Added IsIdle flag to the SmartThreadPool and IWorkItemsGroup
- // - Added support for WinCE
- // - Added support for Action<T> and Func<T>
- //
- // 07 April 2009 - Changes:
- // - Added support for Silverlight and Mono
- // - Added Join, Choice, and Pipe to SmartThreadPool.
- // - Added local performance counters (for Mono, Silverlight, and WindowsCE)
- // - Changed duration measures from DateTime.Now to Stopwatch.
- // - Queues changed from System.Collections.Queue to System.Collections.Generic.LinkedList<T>.
- //
- // 21 December 2009 - Changes:
- // - Added work item timeout (passive)
- //
- // 20 August 2012 - Changes:
- // - Added set name to threads
- // - Fixed the WorkItemsQueue.Dequeue.
- // Replaced while (!Monitor.TryEnter(this)); with lock(this) { ... }
- // - Fixed SmartThreadPool.Pipe
- // - Added IsBackground option to threads
- // - Added ApartmentState to threads
- // - Fixed thread creation when queuing many work items at the same time.
- //
- // 24 August 2012 - Changes:
- // - Enabled cancel abort after cancel. See: http://smartthreadpool.codeplex.com/discussions/345937 by alecswan
- // - Added option to set MaxStackSize of threads
- #endregion
- using System;
- using System.Security;
- using System.Threading;
- using System.Collections;
- using System.Collections.Generic;
- using System.Diagnostics;
- using System.Runtime.CompilerServices;
- using Amib.Threading.Internal;
- namespace Amib.Threading
- {
- #region SmartThreadPool class
- /// <summary>
- /// Smart thread pool class.
- /// </summary>
- public partial class SmartThreadPool : WorkItemsGroupBase, IDisposable
- {
- #region Public Default Constants
- /// <summary>
- /// Default minimum number of threads the thread pool contains. (0)
- /// </summary>
- public const int DefaultMinWorkerThreads = 0;
- /// <summary>
- /// Default maximum number of threads the thread pool contains. (25)
- /// </summary>
- public const int DefaultMaxWorkerThreads = 25;
- /// <summary>
- /// Default idle timeout in milliseconds. (One minute)
- /// </summary>
- public const int DefaultIdleTimeout = 60*1000; // One minute
- /// <summary>
- /// Indicate to copy the security context of the caller and then use it in the call. (false)
- /// </summary>
- public const bool DefaultUseCallerCallContext = false;
- /// <summary>
- /// Indicate to copy the HTTP context of the caller and then use it in the call. (false)
- /// </summary>
- public const bool DefaultUseCallerHttpContext = false;
- /// <summary>
- /// Indicate to dispose of the state objects if they support the IDispose interface. (false)
- /// </summary>
- public const bool DefaultDisposeOfStateObjects = false;
- /// <summary>
- /// The default option to run the post execute (CallToPostExecute.Always)
- /// </summary>
- public const CallToPostExecute DefaultCallToPostExecute = CallToPostExecute.Always;
- /// <summary>
- /// The default post execute method to run. (None)
- /// When null it means not to call it.
- /// </summary>
- public static readonly PostExecuteWorkItemCallback DefaultPostExecuteWorkItemCallback;
- /// <summary>
- /// The default work item priority (WorkItemPriority.Normal)
- /// </summary>
- public const WorkItemPriority DefaultWorkItemPriority = WorkItemPriority.Normal;
- /// <summary>
- /// The default is to work on work items as soon as they arrive
- /// and not to wait for the start. (false)
- /// </summary>
- public const bool DefaultStartSuspended = false;
- /// <summary>
- /// The default name to use for the performance counters instance. (null)
- /// </summary>
- public static readonly string DefaultPerformanceCounterInstanceName;
- #if !(WINDOWS_PHONE)
- /// <summary>
- /// The default thread priority (ThreadPriority.Normal)
- /// </summary>
- public const ThreadPriority DefaultThreadPriority = ThreadPriority.Normal;
- #endif
- /// <summary>
- /// The default thread pool name. (SmartThreadPool)
- /// </summary>
- public const string DefaultThreadPoolName = "SmartThreadPool";
- /// <summary>
- /// The default Max Stack Size. (SmartThreadPool)
- /// </summary>
- public static readonly int? DefaultMaxStackSize = null;
- /// <summary>
- /// The default fill state with params. (false)
- /// It is relevant only to QueueWorkItem of Action<...>/Func<...>
- /// </summary>
- public const bool DefaultFillStateWithArgs = false;
- /// <summary>
- /// The default thread backgroundness. (true)
- /// </summary>
- public const bool DefaultAreThreadsBackground = true;
- #if !(_SILVERLIGHT) && !(WINDOWS_PHONE)
- /// <summary>
- /// The default apartment state of a thread in the thread pool.
- /// The default is ApartmentState.Unknown which means the STP will not
- /// set the apartment of the thread. It will use the .NET default.
- /// </summary>
- public const ApartmentState DefaultApartmentState = ApartmentState.Unknown;
- #endif
- #endregion
- #region Member Variables
- /// <summary>
- /// Dictionary of all the threads in the thread pool.
- /// </summary>
- private readonly SynchronizedDictionary<Thread, ThreadEntry> _workerThreads = new SynchronizedDictionary<Thread, ThreadEntry>();
- /// <summary>
- /// Queue of work items.
- /// </summary>
- private readonly WorkItemsQueue _workItemsQueue = new WorkItemsQueue();
- /// <summary>
- /// Count the work items handled.
- /// Used by the performance counter.
- /// </summary>
- private int _workItemsProcessed;
- /// <summary>
- /// Number of threads that currently work (not idle).
- /// </summary>
- private int _inUseWorkerThreads;
- /// <summary>
- /// Stores a copy of the original STPStartInfo.
- /// It is used to change the MinThread and MaxThreads
- /// </summary>
- private STPStartInfo _stpStartInfo;
- /// <summary>
- /// Total number of work items that are stored in the work items queue
- /// plus the work items that the threads in the pool are working on.
- /// </summary>
- private int _currentWorkItemsCount;
- /// <summary>
- /// Signaled when the thread pool is idle, i.e. no thread is busy
- /// and the work items queue is empty
- /// </summary>
- //private ManualResetEvent _isIdleWaitHandle = new ManualResetEvent(true);
- private ManualResetEvent _isIdleWaitHandle = EventWaitHandleFactory.CreateManualResetEvent(true);
- /// <summary>
- /// An event to signal all the threads to quit immediately.
- /// </summary>
- //private ManualResetEvent _shuttingDownEvent = new ManualResetEvent(false);
- private ManualResetEvent _shuttingDownEvent = EventWaitHandleFactory.CreateManualResetEvent(false);
- /// <summary>
- /// A flag to indicate if the Smart Thread Pool is now suspended.
- /// </summary>
- private bool _isSuspended;
- /// <summary>
- /// A flag to indicate the threads to quit.
- /// </summary>
- private bool _shutdown;
- /// <summary>
- /// Counts the threads created in the pool.
- /// It is used to name the threads.
- /// </summary>
- private int _threadCounter;
- /// <summary>
- /// Indicate that the SmartThreadPool has been disposed
- /// </summary>
- private bool _isDisposed;
- /// <summary>
- /// Holds all the WorkItemsGroup instaces that have at least one
- /// work item int the SmartThreadPool
- /// This variable is used in case of Shutdown
- /// </summary>
- private readonly SynchronizedDictionary<IWorkItemsGroup, IWorkItemsGroup> _workItemsGroups = new SynchronizedDictionary<IWorkItemsGroup, IWorkItemsGroup>();
- /// <summary>
- /// A common object for all the work items int the STP
- /// so we can mark them to cancel in O(1)
- /// </summary>
- private CanceledWorkItemsGroup _canceledSmartThreadPool = new CanceledWorkItemsGroup();
- /// <summary>
- /// Windows STP performance counters
- /// </summary>
- private ISTPInstancePerformanceCounters _windowsPCs = NullSTPInstancePerformanceCounters.Instance;
- /// <summary>
- /// Local STP performance counters
- /// </summary>
- private ISTPInstancePerformanceCounters _localPCs = NullSTPInstancePerformanceCounters.Instance;
- #if (WINDOWS_PHONE)
- private static readonly Dictionary<int, ThreadEntry> _threadEntries = new Dictionary<int, ThreadEntry>();
- #elif (_WINDOWS_CE)
- private static LocalDataStoreSlot _threadEntrySlot = Thread.AllocateDataSlot();
- #else
- [ThreadStatic]
- private static ThreadEntry _threadEntry;
- #endif
- /// <summary>
- /// An event to call after a thread is created, but before
- /// it's first use.
- /// </summary>
- private event ThreadInitializationHandler _onThreadInitialization;
- /// <summary>
- /// An event to call when a thread is about to exit, after
- /// it is no longer belong to the pool.
- /// </summary>
- private event ThreadTerminationHandler _onThreadTermination;
- #endregion
- #region Per thread properties
- /// <summary>
- /// A reference to the current work item a thread from the thread pool
- /// is executing.
- /// </summary>
- internal static ThreadEntry CurrentThreadEntry
- {
- #if (WINDOWS_PHONE)
- get
- {
- lock(_threadEntries)
- {
- ThreadEntry threadEntry;
- if (_threadEntries.TryGetValue(Thread.CurrentThread.ManagedThreadId, out threadEntry))
- {
- return threadEntry;
- }
- }
- return null;
- }
- set
- {
- lock(_threadEntries)
- {
- _threadEntries[Thread.CurrentThread.ManagedThreadId] = value;
- }
- }
- #elif (_WINDOWS_CE)
- get
- {
- //Thread.CurrentThread.ManagedThreadId
- return Thread.GetData(_threadEntrySlot) as ThreadEntry;
- }
- set
- {
- Thread.SetData(_threadEntrySlot, value);
- }
- #else
- get
- {
- return _threadEntry;
- }
- set
- {
- _threadEntry = value;
- }
- #endif
- }
- #endregion
- #region Construction and Finalization
- /// <summary>
- /// Constructor
- /// </summary>
- public SmartThreadPool()
- {
- _stpStartInfo = new STPStartInfo();
- Initialize();
- }
- /// <summary>
- /// Constructor
- /// </summary>
- /// <param name="idleTimeout">Idle timeout in milliseconds</param>
- public SmartThreadPool(int idleTimeout)
- {
- _stpStartInfo = new STPStartInfo
- {
- IdleTimeout = idleTimeout,
- };
- Initialize();
- }
- /// <summary>
- /// Constructor
- /// </summary>
- /// <param name="idleTimeout">Idle timeout in milliseconds</param>
- /// <param name="maxWorkerThreads">Upper limit of threads in the pool</param>
- public SmartThreadPool(
- int idleTimeout,
- int maxWorkerThreads)
- {
- _stpStartInfo = new STPStartInfo
- {
- IdleTimeout = idleTimeout,
- MaxWorkerThreads = maxWorkerThreads,
- };
- Initialize();
- }
- /// <summary>
- /// Constructor
- /// </summary>
- /// <param name="idleTimeout">Idle timeout in milliseconds</param>
- /// <param name="maxWorkerThreads">Upper limit of threads in the pool</param>
- /// <param name="minWorkerThreads">Lower limit of threads in the pool</param>
- public SmartThreadPool(
- int idleTimeout,
- int maxWorkerThreads,
- int minWorkerThreads)
- {
- _stpStartInfo = new STPStartInfo
- {
- IdleTimeout = idleTimeout,
- MaxWorkerThreads = maxWorkerThreads,
- MinWorkerThreads = minWorkerThreads,
- };
- Initialize();
- }
- /// <summary>
- /// Constructor
- /// </summary>
- /// <param name="stpStartInfo">A SmartThreadPool configuration that overrides the default behavior</param>
- public SmartThreadPool(STPStartInfo stpStartInfo)
- {
- _stpStartInfo = new STPStartInfo(stpStartInfo);
- Initialize();
- }
- private void Initialize()
- {
- Name = _stpStartInfo.ThreadPoolName;
- ValidateSTPStartInfo();
- // _stpStartInfoRW stores a read/write copy of the STPStartInfo.
- // Actually only MaxWorkerThreads and MinWorkerThreads are overwritten
- _isSuspended = _stpStartInfo.StartSuspended;
- #if (_WINDOWS_CE) || (_SILVERLIGHT) || (_MONO) || (WINDOWS_PHONE)
- if (null != _stpStartInfo.PerformanceCounterInstanceName)
- {
- throw new NotSupportedException("Performance counters are not implemented for Compact Framework/Silverlight/Mono, instead use StpStartInfo.EnableLocalPerformanceCounters");
- }
- #else
- if (null != _stpStartInfo.PerformanceCounterInstanceName)
- {
- try
- {
- _windowsPCs = new STPInstancePerformanceCounters(_stpStartInfo.PerformanceCounterInstanceName);
- }
- catch (Exception e)
- {
- Debug.WriteLine("Unable to create Performance Counters: " + e);
- _windowsPCs = NullSTPInstancePerformanceCounters.Instance;
- }
- }
- #endif
- if (_stpStartInfo.EnableLocalPerformanceCounters)
- {
- _localPCs = new LocalSTPInstancePerformanceCounters();
- }
- // If the STP is not started suspended then start the threads.
- if (!_isSuspended)
- {
- StartOptimalNumberOfThreads();
- }
- }
- private void StartOptimalNumberOfThreads()
- {
- int threadsCount = Math.Max(_workItemsQueue.Count, _stpStartInfo.MinWorkerThreads);
- threadsCount = Math.Min(threadsCount, _stpStartInfo.MaxWorkerThreads);
- threadsCount -= _workerThreads.Count;
- if (threadsCount > 0)
- {
- StartThreads(threadsCount);
- }
- }
- private void ValidateSTPStartInfo()
- {
- if (_stpStartInfo.MinWorkerThreads < 0)
- {
- throw new ArgumentOutOfRangeException(
- "MinWorkerThreads", "MinWorkerThreads cannot be negative");
- }
- if (_stpStartInfo.MaxWorkerThreads <= 0)
- {
- throw new ArgumentOutOfRangeException(
- "MaxWorkerThreads", "MaxWorkerThreads must be greater than zero");
- }
- if (_stpStartInfo.MinWorkerThreads > _stpStartInfo.MaxWorkerThreads)
- {
- throw new ArgumentOutOfRangeException(
- "MinWorkerThreads, maxWorkerThreads",
- "MaxWorkerThreads must be greater or equal to MinWorkerThreads");
- }
- }
- private static void ValidateCallback(Delegate callback)
- {
- if(callback.GetInvocationList().Length > 1)
- {
- throw new NotSupportedException("SmartThreadPool doesn't support delegates chains");
- }
- }
- #endregion
- #region Thread Processing
- /// <summary>
- /// Waits on the queue for a work item, shutdown, or timeout.
- /// </summary>
- /// <returns>
- /// Returns the WaitingCallback or null in case of timeout or shutdown.
- /// </returns>
- private WorkItem Dequeue()
- {
- WorkItem workItem =
- _workItemsQueue.DequeueWorkItem(_stpStartInfo.IdleTimeout, _shuttingDownEvent);
- return workItem;
- }
- /// <summary>
- /// Put a new work item in the queue
- /// </summary>
- /// <param name="workItem">A work item to queue</param>
- internal override void Enqueue(WorkItem workItem)
- {
- // Make sure the workItem is not null
- Debug.Assert(null != workItem);
- IncrementWorkItemsCount();
- workItem.CanceledSmartThreadPool = _canceledSmartThreadPool;
- _workItemsQueue.EnqueueWorkItem(workItem);
- workItem.WorkItemIsQueued();
- // If all the threads are busy then try to create a new one
- if (_currentWorkItemsCount > _workerThreads.Count)
- {
- StartThreads(1);
- }
- }
- private void IncrementWorkItemsCount()
- {
- _windowsPCs.SampleWorkItems(_workItemsQueue.Count, _workItemsProcessed);
- _localPCs.SampleWorkItems(_workItemsQueue.Count, _workItemsProcessed);
- int count = Interlocked.Increment(ref _currentWorkItemsCount);
- //Trace.WriteLine("WorkItemsCount = " + _currentWorkItemsCount.ToString());
- if (count == 1)
- {
- IsIdle = false;
- _isIdleWaitHandle.Reset();
- }
- }
- private void DecrementWorkItemsCount()
- {
- int count = Interlocked.Decrement(ref _currentWorkItemsCount);
- //Trace.WriteLine("WorkItemsCount = " + _currentWorkItemsCount.ToString());
- if (count == 0)
- {
- IsIdle = true;
- _isIdleWaitHandle.Set();
- }
- Interlocked.Increment(ref _workItemsProcessed);
- if (!_shutdown)
- {
- // The counter counts even if the work item was cancelled
- _windowsPCs.SampleWorkItems(_workItemsQueue.Count, _workItemsProcessed);
- _localPCs.SampleWorkItems(_workItemsQueue.Count, _workItemsProcessed);
- }
- }
- internal void RegisterWorkItemsGroup(IWorkItemsGroup workItemsGroup)
- {
- _workItemsGroups[workItemsGroup] = workItemsGroup;
- }
- internal void UnregisterWorkItemsGroup(IWorkItemsGroup workItemsGroup)
- {
- if (_workItemsGroups.Contains(workItemsGroup))
- {
- _workItemsGroups.Remove(workItemsGroup);
- }
- }
- /// <summary>
- /// Inform that the current thread is about to quit or quiting.
- /// The same thread may call this method more than once.
- /// </summary>
- private void InformCompleted()
- {
- // There is no need to lock the two methods together
- // since only the current thread removes itself
- // and the _workerThreads is a synchronized dictionary
- if (_workerThreads.Contains(Thread.CurrentThread))
- {
- _workerThreads.Remove(Thread.CurrentThread);
- _windowsPCs.SampleThreads(_workerThreads.Count, _inUseWorkerThreads);
- _localPCs.SampleThreads(_workerThreads.Count, _inUseWorkerThreads);
- }
- }
- /// <summary>
- /// Starts new threads
- /// </summary>
- /// <param name="threadsCount">The number of threads to start</param>
- private void StartThreads(int threadsCount)
- {
- if (_isSuspended)
- {
- return;
- }
- lock(_workerThreads.SyncRoot)
- {
- // Don't start threads on shut down
- if (_shutdown)
- {
- return;
- }
- for(int i = 0; i < threadsCount; ++i)
- {
- // Don't create more threads then the upper limit
- if (_workerThreads.Count >= _stpStartInfo.MaxWorkerThreads)
- {
- return;
- }
- // Create a new thread
- #if (_SILVERLIGHT) || (WINDOWS_PHONE)
- Thread workerThread = new Thread(ProcessQueuedItems);
- #else
- Thread workerThread =
- _stpStartInfo.MaxStackSize.HasValue
- ? new Thread(ProcessQueuedItems, _stpStartInfo.MaxStackSize.Value)
- : new Thread(ProcessQueuedItems);
- #endif
- // Configure the new thread and start it
- workerThread.IsBackground = _stpStartInfo.AreThreadsBackground;
- #if !(_SILVERLIGHT) && !(_WINDOWS_CE) && !(WINDOWS_PHONE)
- if (_stpStartInfo.ApartmentState != ApartmentState.Unknown)
- {
- workerThread.SetApartmentState(_stpStartInfo.ApartmentState);
- }
- #endif
- #if !(_SILVERLIGHT) && !(WINDOWS_PHONE)
- workerThread.Priority = _stpStartInfo.ThreadPriority;
- #endif
- workerThread.Name = string.Format("STP:{0}:{1}", Name, _threadCounter);
- workerThread.Start();
- ++_threadCounter;
- // Add it to the dictionary and update its creation time.
- _workerThreads[workerThread] = new ThreadEntry(this);
- _windowsPCs.SampleThreads(_workerThreads.Count, _inUseWorkerThreads);
- _localPCs.SampleThreads(_workerThreads.Count, _inUseWorkerThreads);
- }
- }
- }
- /// <summary>
- /// A worker thread method that processes work items from the work items queue.
- /// </summary>
- private void ProcessQueuedItems()
- {
- // Keep the entry of the dictionary as thread's variable to avoid the synchronization locks
- // of the dictionary.
- CurrentThreadEntry = _workerThreads[Thread.CurrentThread];
- FireOnThreadInitialization();
- try
- {
- bool bInUseWorkerThreadsWasIncremented = false;
- // Process until shutdown.
- while(!_shutdown)
- {
- // Update the last time this thread was seen alive.
- // It's good for debugging.
- CurrentThreadEntry.IAmAlive();
- // The following block handles the when the MaxWorkerThreads has been
- // incremented by the user at run-time.
- // Double lock for quit.
- if (_workerThreads.Count > _stpStartInfo.MaxWorkerThreads)
- {
- lock (_workerThreads.SyncRoot)
- {
- if (_workerThreads.Count > _stpStartInfo.MaxWorkerThreads)
- {
- // Inform that the thread is quiting and then quit.
- // This method must be called within this lock or else
- // more threads will quit and the thread pool will go
- // below the lower limit.
- InformCompleted();
- break;
- }
- }
- }
- // Wait for a work item, shutdown, or timeout
- WorkItem workItem = Dequeue();
- // Update the last time this thread was seen alive.
- // It's good for debugging.
- CurrentThreadEntry.IAmAlive();
- // On timeout or shut down.
- if (null == workItem)
- {
- // Double lock for quit.
- if (_workerThreads.Count > _stpStartInfo.MinWorkerThreads)
- {
- lock(_workerThreads.SyncRoot)
- {
- if (_workerThreads.Count > _stpStartInfo.MinWorkerThreads)
- {
- // Inform that the thread is quiting and then quit.
- // This method must be called within this lock or else
- // more threads will quit and the thread pool will go
- // below the lower limit.
- InformCompleted();
- break;
- }
- }
- }
- }
- // If we didn't quit then skip to the next iteration.
- if (null == workItem)
- {
- continue;
- }
- try
- {
- // Initialize the value to false
- bInUseWorkerThreadsWasIncremented = false;
- // Set the Current Work Item of the thread.
- // Store the Current Work Item before the workItem.StartingWorkItem() is called,
- // so WorkItem.Cancel can work when the work item is between InQueue and InProgress
- // states.
- // If the work item has been cancelled BEFORE the workItem.StartingWorkItem()
- // (work item is in InQueue state) then workItem.StartingWorkItem() will return false.
- // If the work item has been cancelled AFTER the workItem.StartingWorkItem() then
- // (work item is in InProgress state) then the thread will be aborted
- CurrentThreadEntry.CurrentWorkItem = workItem;
- // Change the state of the work item to 'in progress' if possible.
- // We do it here so if the work item has been canceled we won't
- // increment the _inUseWorkerThreads.
- // The cancel mechanism doesn't delete items from the queue,
- // it marks the work item as canceled, and when the work item
- // is dequeued, we just skip it.
- // If the post execute of work item is set to always or to
- // call when the work item is canceled then the StartingWorkItem()
- // will return true, so the post execute can run.
- if (!workItem.StartingWorkItem())
- {
- continue;
- }
- // Execute the callback. Make sure to accurately
- // record how many callbacks are currently executing.
- int inUseWorkerThreads = Interlocked.Increment(ref _inUseWorkerThreads);
- _windowsPCs.SampleThreads(_workerThreads.Count, inUseWorkerThreads);
- _localPCs.SampleThreads(_workerThreads.Count, inUseWorkerThreads);
- // Mark that the _inUseWorkerThreads incremented, so in the finally{}
- // statement we will decrement it correctly.
- bInUseWorkerThreadsWasIncremented = true;
- workItem.FireWorkItemStarted();
- ExecuteWorkItem(workItem);
- }
- catch(Exception ex)
- {
- ex.GetHashCode();
- // Do nothing
- }
- finally
- {
- workItem.DisposeOfState();
- // Set the CurrentWorkItem to null, since we
- // no longer run user's code.
- CurrentThreadEntry.CurrentWorkItem = null;
- // Decrement the _inUseWorkerThreads only if we had
- // incremented it. Note the cancelled work items don't
- // increment _inUseWorkerThreads.
- if (bInUseWorkerThreadsWasIncremented)
- {
- int inUseWorkerThreads = Interlocked.Decrement(ref _inUseWorkerThreads);
- _windowsPCs.SampleThreads(_workerThreads.Count, inUseWorkerThreads);
- _localPCs.SampleThreads(_workerThreads.Count, inUseWorkerThreads);
- }
- // Notify that the work item has been completed.
- // WorkItemsGroup may enqueue their next work item.
- workItem.FireWorkItemCompleted();
- // Decrement the number of work items here so the idle
- // ManualResetEvent won't fluctuate.
- DecrementWorkItemsCount();
- }
- }
- }
- catch(ThreadAbortException tae)
- {
- tae.GetHashCode();
- // Handle the abort exception gracfully.
- #if !(_WINDOWS_CE) && !(_SILVERLIGHT) && !(WINDOWS_PHONE)
- Thread.ResetAbort();
- #endif
- }
- catch(Exception e)
- {
- Debug.Assert(null != e);
- }
- finally
- {
- InformCompleted();
- FireOnThreadTermination();
- }
- }
- private void ExecuteWorkItem(WorkItem workItem)
- {
- _windowsPCs.SampleWorkItemsWaitTime(workItem.WaitingTime);
- _localPCs.SampleWorkItemsWaitTime(workItem.WaitingTime);
- try
- {
- workItem.Execute();
- }
- finally
- {
- _windowsPCs.SampleWorkItemsProcessTime(workItem.ProcessTime);
- _localPCs.SampleWorkItemsProcessTime(workItem.ProcessTime);
- }
- }
- #endregion
- #region Public Methods
- private void ValidateWaitForIdle()
- {
- if (null != CurrentThreadEntry && CurrentThreadEntry.AssociatedSmartThreadPool == this)
- {
- throw new NotSupportedException(
- "WaitForIdle cannot be called from a thread on its SmartThreadPool, it causes a deadlock");
- }
- }
- internal static void ValidateWorkItemsGroupWaitForIdle(IWorkItemsGroup workItemsGroup)
- {
- if (null == CurrentThreadEntry)
- {
- return;
- }
- WorkItem workItem = CurrentThreadEntry.CurrentWorkItem;
- ValidateWorkItemsGroupWaitForIdleImpl(workItemsGroup, workItem);
- if ((null != workItemsGroup) &&
- (null != workItem) &&
- CurrentThreadEntry.CurrentWorkItem.WasQueuedBy(workItemsGroup))
- {
- throw new NotSupportedException("WaitForIdle cannot be called from a thread on its SmartThreadPool, it causes a deadlock");
- }
- }
- [MethodImpl(MethodImplOptions.NoInlining)]
- private static void ValidateWorkItemsGroupWaitForIdleImpl(IWorkItemsGroup workItemsGroup, WorkItem workItem)
- {
- if ((null != workItemsGroup) &&
- (null != workItem) &&
- workItem.WasQueuedBy(workItemsGroup))
- {
- throw new NotSupportedException("WaitForIdle cannot be called from a thread on its SmartThreadPool, it causes a deadlock");
- }
- }
- /// <summary>
- /// Force the SmartThreadPool to shutdown
- /// </summary>
- public void Shutdown()
- {
- Shutdown(true, 0);
- }
- /// <summary>
- /// Force the SmartThreadPool to shutdown with timeout
- /// </summary>
- public void Shutdown(bool forceAbort, TimeSpan timeout)
- {
- Shutdown(forceAbort, (int)timeout.TotalMilliseconds);
- }
- /// <summary>
- /// Empties the queue of work items and abort the threads in the pool.
- /// </summary>
- public void Shutdown(bool forceAbort, int millisecondsTimeout)
- {
- ValidateNotDisposed();
- ISTPInstancePerformanceCounters pcs = _windowsPCs;
- if (NullSTPInstancePerformanceCounters.Instance != _windowsPCs)
- {
- // Set the _pcs to "null" to stop updating the performance
- // counters
- _windowsPCs = NullSTPInstancePerformanceCounters.Instance;
- pcs.Dispose();
- }
- Thread [] threads;
- lock(_workerThreads.SyncRoot)
- {
- // Shutdown the work items queue
- _workItemsQueue.Dispose();
- // Signal the threads to exit
- _shutdown = true;
- _shuttingDownEvent.Set();
- // Make a copy of the threads' references in the pool
- threads = new Thread [_workerThreads.Count];
- _workerThreads.Keys.CopyTo(threads, 0);
- }
- int millisecondsLeft = millisecondsTimeout;
- Stopwatch stopwatch = Stopwatch.StartNew();
- //DateTime start = DateTime.UtcNow;
- bool waitInfinitely = (Timeout.Infinite == millisecondsTimeout);
- bool timeout = false;
- // Each iteration we update the time left for the timeout.
- foreach(Thread thread in threads)
- {
- // Join don't work with negative numbers
- if (!waitInfinitely && (millisecondsLeft < 0))
- {
- timeout = true;
- break;
- }
- // Wait for the thread to terminate
- bool success = thread.Join(millisecondsLeft);
- if(!success)
- {
- timeout = true;
- break;
- }
- if(!waitInfinitely)
- {
- // Update the time left to wait
- //TimeSpan ts = DateTime.UtcNow - start;
- millisecondsLeft = millisecondsTimeout - (int)stopwatch.ElapsedMilliseconds;
- }
- }
- if (timeout && forceAbort)
- {
- // Abort the threads in the pool
- foreach(Thread thread in threads)
- {
-
- if ((thread != null)
- #if !(_WINDOWS_CE)
- && thread.IsAlive
- #endif
- )
- {
- try
- {
- thread.Abort(); // Shutdown
- }
- catch(SecurityException e)
- {
- e.GetHashCode();
- }
- catch(ThreadStateException ex)
- {
- ex.GetHashCode();
- // In case the thread has been terminated
- // after the check if it is alive.
- }
- }
- }
- }
- }
- /// <summary>
- /// Wait for all work items to complete
- /// </summary>
- /// <param name="waitableResults">Array of work item result objects</param>
- /// <returns>
- /// true when every work item in workItemResults has completed; otherwise false.
- /// </returns>
- public static bool WaitAll(
- IWaitableResult [] waitableResults)
- {
- return WaitAll(waitableResults, Timeout.Infinite, true);
- }
- /// <summary>
- /// Wait for all work items to complete
- /// </summary>
- /// <param name="waitableResults">Array of work item result objects</param>
- /// <param name="timeout">The number of milliseconds to wait, or a TimeSpan that represents -1 milliseconds to wait indefinitely. </param>
- /// <param name="exitContext">
- /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
- /// </param>
- /// <returns>
- /// true when every work item in workItemResults has completed; otherwise false.
- /// </returns>
- public static bool WaitAll(
- IWaitableResult [] waitableResults,
- TimeSpan timeout,
- bool exitContext)
- {
- return WaitAll(waitableResults, (int)timeout.TotalMilliseconds, exitContext);
- }
- /// <summary>
- /// Wait for all work items to complete
- /// </summary>
- /// <param name="waitableResults">Array of work item result objects</param>
- /// <param name="timeout">The number of milliseconds to wait, or a TimeSpan that represents -1 milliseconds to wait indefinitely. </param>
- /// <param name="exitContext">
- /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
- /// </param>
- /// <param name="cancelWaitHandle">A cancel wait handle to interrupt the wait if needed</param>
- /// <returns>
- /// true when every work item in workItemResults has completed; otherwise false.
- /// </returns>
- public static bool WaitAll(
- IWaitableResult[] waitableResults,
- TimeSpan timeout,
- bool exitContext,
- WaitHandle cancelWaitHandle)
- {
- return WaitAll(waitableResults, (int)timeout.TotalMilliseconds, exitContext, cancelWaitHandle);
- }
- /// <summary>
- /// Wait for all work items to complete
- /// </summary>
- /// <param name="waitableResults">Array of work item result objects</param>
- /// <param name="millisecondsTimeout">The number of milliseconds to wait, or Timeout.Infinite (-1) to wait indefinitely.</param>
- /// <param name="exitContext">
- /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
- /// </param>
- /// <returns>
- /// true when every work item in workItemResults has completed; otherwise false.
- /// </returns>
- public static bool WaitAll(
- IWaitableResult [] waitableResults,
- int millisecondsTimeout,
- bool exitContext)
- {
- return WorkItem.WaitAll(waitableResults, millisecondsTimeout, exitContext, null);
- }
- /// <summary>
- /// Wait for all work items to complete
- /// </summary>
- /// <param name="waitableResults">Array of work item result objects</param>
- /// <param name="millisecondsTimeout">The number of milliseconds to wait, or Timeout.Infinite (-1) to wait indefinitely.</param>
- /// <param name="exitContext">
- /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
- /// </param>
- /// <param name="cancelWaitHandle">A cancel wait handle to interrupt the wait if needed</param>
- /// <returns>
- /// true when every work item in workItemResults has completed; otherwise false.
- /// </returns>
- public static bool WaitAll(
- IWaitableResult[] waitableResults,
- int millisecondsTimeout,
- bool exitContext,
- WaitHandle cancelWaitHandle)
- {
- return WorkItem.WaitAll(waitableResults, millisecondsTimeout, exitContext, cancelWaitHandle);
- }
- /// <summary>
- /// Waits for any of the work items in the specified array to complete, cancel, or timeout
- /// </summary>
- /// <param name="waitableResults">Array of work item result objects</param>
- /// <returns>
- /// The array index of the work item result that satisfied the wait, or WaitTimeout if any of the work items has been canceled.
- /// </returns>
- public static int WaitAny(
- IWaitableResult [] waitableResults)
- {
- return WaitAny(waitableResults, Timeout.Infinite, true);
- }
- /// <summary>
- /// Waits for any of the work items in the specified array to complete, cancel, or timeout
- /// </summary>
- /// <param name="waitableResults">Array of work item result objects</param>
- /// <param name="timeout">The number of milliseconds to wait, or a TimeSpan that represents -1 milliseconds to wait indefinitely. </param>
- /// <param name="exitContext">
- /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
- /// </param>
- /// <returns>
- /// The array index of the work item result that satisfied the wait, or WaitTimeout if no work item result satisfied the wait and a time interval equivalent to millisecondsTimeout has passed or the work item has been canceled.
- /// </returns>
- public static int WaitAny(
- IWaitableResult[] waitableResults,
- TimeSpan timeout,
- bool exitContext)
- {
- return WaitAny(waitableResults, (int)timeout.TotalMilliseconds, exitContext);
- }
- /// <summary>
- /// Waits for any of the work items in the specified array to complete, cancel, or timeout
- /// </summary>
- /// <param name="waitableResults">Array of work item result objects</param>
- /// <param name="timeout">The number of milliseconds to wait, or a TimeSpan that represents -1 milliseconds to wait indefinitely. </param>
- /// <param name="exitContext">
- /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
- /// </param>
- /// <param name="cancelWaitHandle">A cancel wait handle to interrupt the wait if needed</param>
- /// <returns>
- /// The array index of the work item result that satisfied the wait, or WaitTimeout if no work item result satisfied the wait and a time interval equivalent to millisecondsTimeout has passed or the work item has been canceled.
- /// </returns>
- public static int WaitAny(
- IWaitableResult [] waitableResults,
- TimeSpan timeout,
- bool exitContext,
- WaitHandle cancelWaitHandle)
- {
- return WaitAny(waitableResults, (int)timeout.TotalMilliseconds, exitContext, cancelWaitHandle);
- }
- /// <summary>
- /// Waits for any of the work items in the specified array to complete, cancel, or timeout
- /// </summary>
- /// <param name="waitableResults">Array of work item result objects</param>
- /// <param name="millisecondsTimeout">The number of milliseconds to wait, or Timeout.Infinite (-1) to wait indefinitely.</param>
- /// <param name="exitContext">
- /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
- /// </param>
- /// <returns>
- /// The array index of the work item result that satisfied the wait, or WaitTimeout if no work item result satisfied the wait and a time interval equivalent to millisecondsTimeout has passed or the work item has been canceled.
- /// </returns>
- public static int WaitAny(
- IWaitableResult [] waitableResults,
- int millisecondsTimeout,
- bool exitContext)
- {
- return WorkItem.WaitAny(waitableResults, millisecondsTimeout, exitContext, null);
- }
- /// <summary>
- /// Waits for any of the work items in the specified array to complete, cancel, or timeout
- /// </summary>
- /// <param name="waitableResults">Array of work item result objects</param>
- /// <param name="millisecondsTimeout">The number of milliseconds to wait, or Timeout.Infinite (-1) to wait indefinitely.</param>
- /// <param name="exitContext">
- /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
- /// </param>
- /// <param name="cancelWaitHandle">A cancel wait handle to interrupt the wait if needed</param>
- /// <returns>
- /// The array index of the work item result that satisfied the wait, or WaitTimeout if no work item result satisfied the wait and a time interval equivalent to millisecondsTimeout has passed or the work item has been canceled.
- /// </returns>
- public static int WaitAny(
- IWaitableResult [] waitableResults,
- int millisecondsTimeout,
- bool exitContext,
- WaitHandle cancelWaitHandle)
- {
- return WorkItem.WaitAny(waitableResults, millisecondsTimeout, exitContext, cancelWaitHandle);
- }
- /// <summary>
- /// Creates a new WorkItemsGroup.
- /// </summary>
- /// <param name="concurrency">The number of work items that can be run concurrently</param>
- /// <returns>A reference to the WorkItemsGroup</returns>
- public IWorkItemsGroup CreateWorkItemsGroup(int concurrency)
- {
- IWorkItemsGroup workItemsGroup = new WorkItemsGroup(this, concurrency, _stpStartInfo);
- return workItemsGroup;
- }
- /// <summary>
- /// Creates a new WorkItemsGroup.
- /// </summary>
- /// <param name="concurrency">The number of work items that can be run concurrently</param>
- /// <param name="wigStartInfo">A WorkItemsGroup configuration that overrides the default behavior</param>
- /// <returns>A reference to the WorkItemsGroup</returns>
- public IWorkItemsGroup CreateWorkItemsGroup(int concurrency, WIGStartInfo wigStartInfo)
- {
- IWorkItemsGroup workItemsGroup = new WorkItemsGroup(this, concurrency, wigStartInfo);
- return workItemsGroup;
- }
- #region Fire Thread's Events
- private void FireOnThreadInitialization()
- {
- if (null != _onThreadInitialization)
- {
- foreach (ThreadInitializationHandler tih in _onThreadInitialization.GetInvocationList())
- {
- try
- {
- tih();
- }
- catch (Exception e)
- {
- e.GetHashCode();
- Debug.Assert(false);
- throw;
- }
- }
- }
- }
- private void FireOnThreadTermination()
- {
- if (null != _onThreadTermination)
- {
- foreach (ThreadTerminationHandler tth in _onThreadTermination.GetInvocationList())
- {
- try
- {
- tth();
- }
- catch (Exception e)
- {
- e.GetHashCode();
- Debug.Assert(false);
- throw;
- }
- }
- }
- }
- #endregion
- /// <summary>
- /// This event is fired when a thread is created.
- /// Use it to initialize a thread before the work items use it.
- /// </summary>
- public event ThreadInitializationHandler OnThreadInitialization
- {
- add { _onThreadInitialization += value; }
- remove { _onThreadInitialization -= value; }
- }
- /// <summary>
- /// This event is fired when a thread is terminating.
- /// Use it for cleanup.
- /// </summary>
- public event ThreadTerminationHandler OnThreadTermination
- {
- add { _onThreadTermination += value; }
- remove { _onThreadTermination -= value; }
- }
- internal void CancelAbortWorkItemsGroup(WorkItemsGroup wig)
- {
- foreach (ThreadEntry threadEntry in _workerThreads.Values)
- {
- WorkItem workItem = threadEntry.CurrentWorkItem;
- if (null != workItem &&
- workItem.WasQueuedBy(wig) &&
- !workItem.IsCanceled)
- {
- threadEntry.CurrentWorkItem.GetWorkItemResult().Cancel(true);
- }
- }
- }
-
- #endregion
- #region Properties
- /// <summary>
- /// Get/Set the lower limit of threads in the pool.
- /// </summary>
- public int MinThreads
- {
- get
- {
- ValidateNotDisposed();
- return _stpStartInfo.MinWorkerThreads;
- }
- set
- {
- Debug.Assert(value >= 0);
- Debug.Assert(value <= _stpStartInfo.MaxWorkerThreads);
- if (_stpStartInfo.MaxWorkerThreads < value)
- {
- _stpStartInfo.MaxWorkerThreads = value;
- }
- _stpStartInfo.MinWorkerThreads = value;
- StartOptimalNumberOfThreads();
- }
- }
- /// <summary>
- /// Get/Set the upper limit of threads in the pool.
- /// </summary>
- public int MaxThreads
- {
- get
- {
- ValidateNotDisposed();
- return _stpStartInfo.MaxWorkerThreads;
- }
- set
- {
- Debug.Assert(value > 0);
- Debug.Assert(value >= _stpStartInfo.MinWorkerThreads);
- if (_stpStartInfo.MinWorkerThreads > value)
- {
- _stpStartInfo.MinWorkerThreads = value;
- }
- _stpStartInfo.MaxWorkerThreads = value;
- StartOptimalNumberOfThreads();
- }
- }
- /// <summary>
- /// Get the number of threads in the thread pool.
- /// Should be between the lower and the upper limits.
- /// </summary>
- public int ActiveThreads
- {
- get
- {
- ValidateNotDisposed();
- return _workerThreads.Count;
- }
- }
- /// <summary>
- /// Get the number of busy (not idle) threads in the thread pool.
- /// </summary>
- public int InUseThreads
- {
- get
- {
- ValidateNotDisposed();
- return _inUseWorkerThreads;
- }
- }
- /// <summary>
- /// Returns true if the current running work item has been cancelled.
- /// Must be used within the work item's callback method.
- /// The work item should sample this value in order to know if it
- /// needs to quit before its completion.
- /// </summary>
- public static bool IsWorkItemCanceled
- {
- get
- {
- return CurrentThreadEntry.CurrentWorkItem.IsCanceled;
- }
- }
-
- /// <summary>
- /// Checks if the work item has been cancelled, and if yes then abort the thread.
- /// Can be used with Cancel and timeout
- /// </summary>
- public static void AbortOnWorkItemCancel()
- {
- if (IsWorkItemCanceled)
- {
- Thread.CurrentThread.Abort();
- }
- }
- /// <summary>
- /// Thread Pool start information (readonly)
- /// </summary>
- public STPStartInfo STPStartInfo
- {
- get
- {
- return _stpStartInfo.AsReadOnly();
- }
- }
- public bool IsShuttingdown
- {
- get { return _shutdown; }
- }
- /// <summary>
- /// Return the local calculated performance counters
- /// Available only if STPStartInfo.EnableLocalPerformanceCounters is true.
- /// </summary>
- public ISTPPerformanceCountersReader PerformanceCountersReader
- {
- get { return (ISTPPerformanceCountersReader)_localPCs; }
- }
- #endregion
- #region IDisposable Members
- public void Dispose()
- {
- if (!_isDisposed)
- {
- if (!_shutdown)
- {
- Shutdown();
- }
- if (null != _shuttingDownEvent)
- {
- _shuttingDownEvent.Close();
- _shuttingDownEvent = null;
- }
- _workerThreads.Clear();
-
- if (null != _isIdleWaitHandle)
- {
- _isIdleWaitHandle.Close();
- _isIdleWaitHandle = null;
- }
- if (_stpStartInfo.EnableLocalPerformanceCounters)
- _localPCs.Dispose();
- _isDisposed = true;
- }
- }
- private void ValidateNotDisposed()
- {
- if(_isDisposed)
- {
- throw new ObjectDisposedException(GetType().ToString(), "The SmartThreadPool has been shutdown");
- }
- }
- #endregion
- #region WorkItemsGroupBase Overrides
- /// <summary>
- /// Get/Set the maximum number of work items that execute cocurrency on the thread pool
- /// </summary>
- public override int Concurrency
- {
- get { return MaxThreads; }
- set { MaxThreads = value; }
- }
- /// <summary>
- /// Get the number of work items in the queue.
- /// </summary>
- public override int WaitingCallbacks
- {
- get
- {
- ValidateNotDisposed();
- return _workItemsQueue.Count;
- }
- }
- /// <summary>
- /// Get an array with all the state objects of the currently running items.
- /// The array represents a snap shot and impact performance.
- /// </summary>
- public override object[] GetStates()
- {
- object[] states = _workItemsQueue.GetStates();
- return states;
- }
- /// <summary>
- /// WorkItemsGroup start information (readonly)
- /// </summary>
- public override WIGStartInfo WIGStartInfo
- {
- get { return _stpStartInfo.AsReadOnly(); }
- }
- /// <summary>
- /// Start the thread pool if it was started suspended.
- /// If it is already running, this method is ignored.
- /// </summary>
- public override void Start()
- {
- if (!_isSuspended)
- {
- return;
- }
- _isSuspended = false;
- ICollection workItemsGroups = _workItemsGroups.Values;
- foreach (WorkItemsGroup workItemsGroup in workItemsGroups)
- {
- workItemsGroup.OnSTPIsStarting();
- }
- StartOptimalNumberOfThreads();
- }
- /// <summary>
- /// Cancel all work items using thread abortion
- /// </summary>
- /// <param name="abortExecution">True to stop work items by raising ThreadAbortException</param>
- public override void Cancel(bool abortExecution)
- {
- _canceledSmartThreadPool.IsCanceled = true;
- _canceledSmartThreadPool = new CanceledWorkItemsGroup();
- ICollection workItemsGroups = _workItemsGroups.Values;
- foreach (WorkItemsGroup workItemsGroup in workItemsGroups)
- {
- workItemsGroup.Cancel(abortExecution);
- }
- if (abortExecution)
- {
- foreach (ThreadEntry threadEntry in _workerThreads.Values)
- {
- WorkItem workItem = threadEntry.CurrentWorkItem;
- if (null != workItem &&
- threadEntry.AssociatedSmartThreadPool == this &&
- !workItem.IsCanceled)
- {
- threadEntry.CurrentWorkItem.GetWorkItemResult().Cancel(true);
- }
- }
- }
- }
- /// <summary>
- /// Wait for the thread pool to be idle
- /// </summary>
- public override bool WaitForIdle(int millisecondsTimeout)
- {
- ValidateWaitForIdle();
- return STPEventWaitHandle.WaitOne(_isIdleWaitHandle, millisecondsTimeout, false);
- }
- /// <summary>
- /// This event is fired when all work items are completed.
- /// (When IsIdle changes to true)
- /// This event only work on WorkItemsGroup. On SmartThreadPool
- /// it throws the NotImplementedException.
- /// </summary>
- public override event WorkItemsGroupIdleHandler OnIdle
- {
- add
- {
- throw new NotImplementedException("This event is not implemented in the SmartThreadPool class. Please create a WorkItemsGroup in order to use this feature.");
- //_onIdle += value;
- }
- remove
- {
- throw new NotImplementedException("This event is not implemented in the SmartThreadPool class. Please create a WorkItemsGroup in order to use this feature.");
- //_onIdle -= value;
- }
- }
- internal override void PreQueueWorkItem()
- {
- ValidateNotDisposed();
- }
- #endregion
- #region Join, Choice, Pipe, etc.
- /// <summary>
- /// Executes all actions in parallel.
- /// Returns when they all finish.
- /// </summary>
- /// <param name="actions">Actions to execute</param>
- public void Join(IEnumerable<Action> actions)
- {
- WIGStartInfo wigStartInfo = new WIGStartInfo { StartSuspended = true };
- IWorkItemsGroup workItemsGroup = CreateWorkItemsGroup(int.MaxValue, wigStartInfo);
- foreach (Action action in actions)
- {
- workItemsGroup.QueueWorkItem(action);
- }
- workItemsGroup.Start();
- workItemsGroup.WaitForIdle();
- }
- /// <summary>
- /// Executes all actions in parallel.
- /// Returns when they all finish.
- /// </summary>
- /// <param name="actions">Actions to execute</param>
- public void Join(params Action[] actions)
- {
- Join((IEnumerable<Action>)actions);
- }
- private class ChoiceIndex
- {
- public int _index = -1;
- }
- /// <summary>
- /// Executes all actions in parallel
- /// Returns when the first one completes
- /// </summary>
- /// <param name="actions">Actions to execute</param>
- public int Choice(IEnumerable<Action> actions)
- {
- WIGStartInfo wigStartInfo = new WIGStartInfo { StartSuspended = true };
- IWorkItemsGroup workItemsGroup = CreateWorkItemsGroup(int.MaxValue, wigStartInfo);
- ManualResetEvent anActionCompleted = new ManualResetEvent(false);
- ChoiceIndex choiceIndex = new ChoiceIndex();
-
- int i = 0;
- foreach (Action action in actions)
- {
- Action act = action;
- int value = i;
- workItemsGroup.QueueWorkItem(() => { act(); Interlocked.CompareExchange(ref choiceIndex._index, value, -1); anActionCompleted.Set(); });
- ++i;
- }
- workItemsGroup.Start();
- anActionCompleted.WaitOne();
- anActionCompleted.Dispose();
- return choiceIndex._index;
- }
- /// <summary>
- /// Executes all actions in parallel
- /// Returns when the first one completes
- /// </summary>
- /// <param name="actions">Actions to execute</param>
- public int Choice(params Action[] actions)
- {
- return Choice((IEnumerable<Action>)actions);
- }
- /// <summary>
- /// Executes actions in sequence asynchronously.
- /// Returns immediately.
- /// </summary>
- /// <param name="pipeState">A state context that passes </param>
- /// <param name="actions">Actions to execute in the order they should run</param>
- public void Pipe<T>(T pipeState, IEnumerable<Action<T>> actions)
- {
- WIGStartInfo wigStartInfo = new WIGStartInfo { StartSuspended = true };
- IWorkItemsGroup workItemsGroup = CreateWorkItemsGroup(1, wigStartInfo);
- foreach (Action<T> action in actions)
- {
- Action<T> act = action;
- workItemsGroup.QueueWorkItem(() => act(pipeState));
- }
- workItemsGroup.Start();
- workItemsGroup.WaitForIdle();
- }
- /// <summary>
- /// Executes actions in sequence asynchronously.
- /// Returns immediately.
- /// </summary>
- /// <param name="pipeState"></param>
- /// <param name="actions">Actions to execute in the order they should run</param>
- public void Pipe<T>(T pipeState, params Action<T>[] actions)
- {
- Pipe(pipeState, (IEnumerable<Action<T>>)actions);
- }
- #endregion
- }
- #endregion
- }
|