SmartThreadPool.cs 64 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361362363364365366367368369370371372373374375376377378379380381382383384385386387388389390391392393394395396397398399400401402403404405406407408409410411412413414415416417418419420421422423424425426427428429430431432433434435436437438439440441442443444445446447448449450451452453454455456457458459460461462463464465466467468469470471472473474475476477478479480481482483484485486487488489490491492493494495496497498499500501502503504505506507508509510511512513514515516517518519520521522523524525526527528529530531532533534535536537538539540541542543544545546547548549550551552553554555556557558559560561562563564565566567568569570571572573574575576577578579580581582583584585586587588589590591592593594595596597598599600601602603604605606607608609610611612613614615616617618619620621622623624625626627628629630631632633634635636637638639640641642643644645646647648649650651652653654655656657658659660661662663664665666667668669670671672673674675676677678679680681682683684685686687688689690691692693694695696697698699700701702703704705706707708709710711712713714715716717718719720721722723724725726727728729730731732733734735736737738739740741742743744745746747748749750751752753754755756757758759760761762763764765766767768769770771772773774775776777778779780781782783784785786787788789790791792793794795796797798799800801802803804805806807808809810811812813814815816817818819820821822823824825826827828829830831832833834835836837838839840841842843844845846847848849850851852853854855856857858859860861862863864865866867868869870871872873874875876877878879880881882883884885886887888889890891892893894895896897898899900901902903904905906907908909910911912913914915916917918919920921922923924925926927928929930931932933934935936937938939940941942943944945946947948949950951952953954955956957958959960961962963964965966967968969970971972973974975976977978979980981982983984985986987988989990991992993994995996997998999100010011002100310041005100610071008100910101011101210131014101510161017101810191020102110221023102410251026102710281029103010311032103310341035103610371038103910401041104210431044104510461047104810491050105110521053105410551056105710581059106010611062106310641065106610671068106910701071107210731074107510761077107810791080108110821083108410851086108710881089109010911092109310941095109610971098109911001101110211031104110511061107110811091110111111121113111411151116111711181119112011211122112311241125112611271128112911301131113211331134113511361137113811391140114111421143114411451146114711481149115011511152115311541155115611571158115911601161116211631164116511661167116811691170117111721173117411751176117711781179118011811182118311841185118611871188118911901191119211931194119511961197119811991200120112021203120412051206120712081209121012111212121312141215121612171218121912201221122212231224122512261227122812291230123112321233123412351236123712381239124012411242124312441245124612471248124912501251125212531254125512561257125812591260126112621263126412651266126712681269127012711272127312741275127612771278127912801281128212831284128512861287128812891290129112921293129412951296129712981299130013011302130313041305130613071308130913101311131213131314131513161317131813191320132113221323132413251326132713281329133013311332133313341335133613371338133913401341134213431344134513461347134813491350135113521353135413551356135713581359136013611362136313641365136613671368136913701371137213731374137513761377137813791380138113821383138413851386138713881389139013911392139313941395139613971398139914001401140214031404140514061407140814091410141114121413141414151416141714181419142014211422142314241425142614271428142914301431143214331434143514361437143814391440144114421443144414451446144714481449145014511452145314541455145614571458145914601461146214631464146514661467146814691470147114721473147414751476147714781479148014811482148314841485148614871488148914901491149214931494149514961497149814991500150115021503150415051506150715081509151015111512151315141515151615171518151915201521152215231524152515261527152815291530153115321533153415351536153715381539154015411542154315441545154615471548154915501551155215531554155515561557155815591560156115621563156415651566156715681569157015711572157315741575157615771578157915801581158215831584158515861587158815891590159115921593159415951596159715981599160016011602160316041605160616071608160916101611161216131614161516161617161816191620162116221623162416251626162716281629163016311632163316341635163616371638163916401641164216431644164516461647164816491650165116521653165416551656165716581659166016611662166316641665166616671668166916701671167216731674167516761677167816791680168116821683168416851686168716881689169016911692169316941695169616971698169917001701170217031704170517061707170817091710171117121713171417151716171717181719172017211722172317241725172617271728172917301731173217331734173517361737
  1. #region Release History
  2. // Smart Thread Pool
  3. // 7 Aug 2004 - Initial release
  4. //
  5. // 14 Sep 2004 - Bug fixes
  6. //
  7. // 15 Oct 2004 - Added new features
  8. // - Work items return result.
  9. // - Support waiting synchronization for multiple work items.
  10. // - Work items can be cancelled.
  11. // - Passage of the caller thread’s context to the thread in the pool.
  12. // - Minimal usage of WIN32 handles.
  13. // - Minor bug fixes.
  14. //
  15. // 26 Dec 2004 - Changes:
  16. // - Removed static constructors.
  17. // - Added finalizers.
  18. // - Changed Exceptions so they are serializable.
  19. // - Fixed the bug in one of the SmartThreadPool constructors.
  20. // - Changed the SmartThreadPool.WaitAll() so it will support any number of waiters.
  21. // The SmartThreadPool.WaitAny() is still limited by the .NET Framework.
  22. // - Added PostExecute with options on which cases to call it.
  23. // - Added option to dispose of the state objects.
  24. // - Added a WaitForIdle() method that waits until the work items queue is empty.
  25. // - Added an STPStartInfo class for the initialization of the thread pool.
  26. // - Changed exception handling so if a work item throws an exception it
  27. // is rethrown at GetResult(), rather then firing an UnhandledException event.
  28. // Note that PostExecute exception are always ignored.
  29. //
  30. // 25 Mar 2005 - Changes:
  31. // - Fixed lost of work items bug
  32. //
  33. // 3 Jul 2005: Changes.
  34. // - Fixed bug where Enqueue() throws an exception because PopWaiter() returned null, hardly reconstructed.
  35. //
  36. // 16 Aug 2005: Changes.
  37. // - Fixed bug where the InUseThreads becomes negative when canceling work items.
  38. //
  39. // 31 Jan 2006 - Changes:
  40. // - Added work items priority
  41. // - Removed support of chained delegates in callbacks and post executes (nobody really use this)
  42. // - Added work items groups
  43. // - Added work items groups idle event
  44. // - Changed SmartThreadPool.WaitAll() behavior so when it gets empty array
  45. // it returns true rather then throwing an exception.
  46. // - Added option to start the STP and the WIG as suspended
  47. // - Exception behavior changed, the real exception is returned by an
  48. // inner exception
  49. // - Added option to keep the Http context of the caller thread. (Thanks to Steven T.)
  50. // - Added performance counters
  51. // - Added priority to the threads in the pool
  52. //
  53. // 13 Feb 2006 - Changes:
  54. // - Added a call to the dispose of the Performance Counter so
  55. // their won't be a Performance Counter leak.
  56. // - Added exception catch in case the Performance Counters cannot
  57. // be created.
  58. //
  59. // 17 May 2008 - Changes:
  60. // - Changed the dispose behavior and removed the Finalizers.
  61. // - Enabled the change of the MaxThreads and MinThreads at run time.
  62. // - Enabled the change of the Concurrency of a IWorkItemsGroup at run
  63. // time If the IWorkItemsGroup is a SmartThreadPool then the Concurrency
  64. // refers to the MaxThreads.
  65. // - Improved the cancel behavior.
  66. // - Added events for thread creation and termination.
  67. // - Fixed the HttpContext context capture.
  68. // - Changed internal collections so they use generic collections
  69. // - Added IsIdle flag to the SmartThreadPool and IWorkItemsGroup
  70. // - Added support for WinCE
  71. // - Added support for Action<T> and Func<T>
  72. //
  73. // 07 April 2009 - Changes:
  74. // - Added support for Silverlight and Mono
  75. // - Added Join, Choice, and Pipe to SmartThreadPool.
  76. // - Added local performance counters (for Mono, Silverlight, and WindowsCE)
  77. // - Changed duration measures from DateTime.Now to Stopwatch.
  78. // - Queues changed from System.Collections.Queue to System.Collections.Generic.LinkedList<T>.
  79. //
  80. // 21 December 2009 - Changes:
  81. // - Added work item timeout (passive)
  82. //
  83. // 20 August 2012 - Changes:
  84. // - Added set name to threads
  85. // - Fixed the WorkItemsQueue.Dequeue.
  86. // Replaced while (!Monitor.TryEnter(this)); with lock(this) { ... }
  87. // - Fixed SmartThreadPool.Pipe
  88. // - Added IsBackground option to threads
  89. // - Added ApartmentState to threads
  90. // - Fixed thread creation when queuing many work items at the same time.
  91. //
  92. // 24 August 2012 - Changes:
  93. // - Enabled cancel abort after cancel. See: http://smartthreadpool.codeplex.com/discussions/345937 by alecswan
  94. // - Added option to set MaxStackSize of threads
  95. #endregion
  96. using System;
  97. using System.Security;
  98. using System.Threading;
  99. using System.Collections;
  100. using System.Collections.Generic;
  101. using System.Diagnostics;
  102. using System.Runtime.CompilerServices;
  103. using Amib.Threading.Internal;
  104. namespace Amib.Threading
  105. {
  106. #region SmartThreadPool class
  107. /// <summary>
  108. /// Smart thread pool class.
  109. /// </summary>
  110. public partial class SmartThreadPool : WorkItemsGroupBase, IDisposable
  111. {
  112. #region Public Default Constants
  113. /// <summary>
  114. /// Default minimum number of threads the thread pool contains. (0)
  115. /// </summary>
  116. public const int DefaultMinWorkerThreads = 0;
  117. /// <summary>
  118. /// Default maximum number of threads the thread pool contains. (25)
  119. /// </summary>
  120. public const int DefaultMaxWorkerThreads = 25;
  121. /// <summary>
  122. /// Default idle timeout in milliseconds. (One minute)
  123. /// </summary>
  124. public const int DefaultIdleTimeout = 60*1000; // One minute
  125. /// <summary>
  126. /// Indicate to copy the security context of the caller and then use it in the call. (false)
  127. /// </summary>
  128. public const bool DefaultUseCallerCallContext = false;
  129. /// <summary>
  130. /// Indicate to copy the HTTP context of the caller and then use it in the call. (false)
  131. /// </summary>
  132. public const bool DefaultUseCallerHttpContext = false;
  133. /// <summary>
  134. /// Indicate to dispose of the state objects if they support the IDispose interface. (false)
  135. /// </summary>
  136. public const bool DefaultDisposeOfStateObjects = false;
  137. /// <summary>
  138. /// The default option to run the post execute (CallToPostExecute.Always)
  139. /// </summary>
  140. public const CallToPostExecute DefaultCallToPostExecute = CallToPostExecute.Always;
  141. /// <summary>
  142. /// The default post execute method to run. (None)
  143. /// When null it means not to call it.
  144. /// </summary>
  145. public static readonly PostExecuteWorkItemCallback DefaultPostExecuteWorkItemCallback;
  146. /// <summary>
  147. /// The default work item priority (WorkItemPriority.Normal)
  148. /// </summary>
  149. public const WorkItemPriority DefaultWorkItemPriority = WorkItemPriority.Normal;
  150. /// <summary>
  151. /// The default is to work on work items as soon as they arrive
  152. /// and not to wait for the start. (false)
  153. /// </summary>
  154. public const bool DefaultStartSuspended = false;
  155. /// <summary>
  156. /// The default name to use for the performance counters instance. (null)
  157. /// </summary>
  158. public static readonly string DefaultPerformanceCounterInstanceName;
  159. #if !(WINDOWS_PHONE)
  160. /// <summary>
  161. /// The default thread priority (ThreadPriority.Normal)
  162. /// </summary>
  163. public const ThreadPriority DefaultThreadPriority = ThreadPriority.Normal;
  164. #endif
  165. /// <summary>
  166. /// The default thread pool name. (SmartThreadPool)
  167. /// </summary>
  168. public const string DefaultThreadPoolName = "SmartThreadPool";
  169. /// <summary>
  170. /// The default Max Stack Size. (SmartThreadPool)
  171. /// </summary>
  172. public static readonly int? DefaultMaxStackSize = null;
  173. /// <summary>
  174. /// The default fill state with params. (false)
  175. /// It is relevant only to QueueWorkItem of Action&lt;...&gt;/Func&lt;...&gt;
  176. /// </summary>
  177. public const bool DefaultFillStateWithArgs = false;
  178. /// <summary>
  179. /// The default thread backgroundness. (true)
  180. /// </summary>
  181. public const bool DefaultAreThreadsBackground = true;
  182. #if !(_SILVERLIGHT) && !(WINDOWS_PHONE)
  183. /// <summary>
  184. /// The default apartment state of a thread in the thread pool.
  185. /// The default is ApartmentState.Unknown which means the STP will not
  186. /// set the apartment of the thread. It will use the .NET default.
  187. /// </summary>
  188. public const ApartmentState DefaultApartmentState = ApartmentState.Unknown;
  189. #endif
  190. #endregion
  191. #region Member Variables
  192. /// <summary>
  193. /// Dictionary of all the threads in the thread pool.
  194. /// </summary>
  195. private readonly SynchronizedDictionary<Thread, ThreadEntry> _workerThreads = new SynchronizedDictionary<Thread, ThreadEntry>();
  196. /// <summary>
  197. /// Queue of work items.
  198. /// </summary>
  199. private readonly WorkItemsQueue _workItemsQueue = new WorkItemsQueue();
  200. /// <summary>
  201. /// Count the work items handled.
  202. /// Used by the performance counter.
  203. /// </summary>
  204. private int _workItemsProcessed;
  205. /// <summary>
  206. /// Number of threads that currently work (not idle).
  207. /// </summary>
  208. private int _inUseWorkerThreads;
  209. /// <summary>
  210. /// Stores a copy of the original STPStartInfo.
  211. /// It is used to change the MinThread and MaxThreads
  212. /// </summary>
  213. private STPStartInfo _stpStartInfo;
  214. /// <summary>
  215. /// Total number of work items that are stored in the work items queue
  216. /// plus the work items that the threads in the pool are working on.
  217. /// </summary>
  218. private int _currentWorkItemsCount;
  219. /// <summary>
  220. /// Signaled when the thread pool is idle, i.e. no thread is busy
  221. /// and the work items queue is empty
  222. /// </summary>
  223. //private ManualResetEvent _isIdleWaitHandle = new ManualResetEvent(true);
  224. private ManualResetEvent _isIdleWaitHandle = EventWaitHandleFactory.CreateManualResetEvent(true);
  225. /// <summary>
  226. /// An event to signal all the threads to quit immediately.
  227. /// </summary>
  228. //private ManualResetEvent _shuttingDownEvent = new ManualResetEvent(false);
  229. private ManualResetEvent _shuttingDownEvent = EventWaitHandleFactory.CreateManualResetEvent(false);
  230. /// <summary>
  231. /// A flag to indicate if the Smart Thread Pool is now suspended.
  232. /// </summary>
  233. private bool _isSuspended;
  234. /// <summary>
  235. /// A flag to indicate the threads to quit.
  236. /// </summary>
  237. private bool _shutdown;
  238. /// <summary>
  239. /// Counts the threads created in the pool.
  240. /// It is used to name the threads.
  241. /// </summary>
  242. private int _threadCounter;
  243. /// <summary>
  244. /// Indicate that the SmartThreadPool has been disposed
  245. /// </summary>
  246. private bool _isDisposed;
  247. /// <summary>
  248. /// Holds all the WorkItemsGroup instaces that have at least one
  249. /// work item int the SmartThreadPool
  250. /// This variable is used in case of Shutdown
  251. /// </summary>
  252. private readonly SynchronizedDictionary<IWorkItemsGroup, IWorkItemsGroup> _workItemsGroups = new SynchronizedDictionary<IWorkItemsGroup, IWorkItemsGroup>();
  253. /// <summary>
  254. /// A common object for all the work items int the STP
  255. /// so we can mark them to cancel in O(1)
  256. /// </summary>
  257. private CanceledWorkItemsGroup _canceledSmartThreadPool = new CanceledWorkItemsGroup();
  258. /// <summary>
  259. /// Windows STP performance counters
  260. /// </summary>
  261. private ISTPInstancePerformanceCounters _windowsPCs = NullSTPInstancePerformanceCounters.Instance;
  262. /// <summary>
  263. /// Local STP performance counters
  264. /// </summary>
  265. private ISTPInstancePerformanceCounters _localPCs = NullSTPInstancePerformanceCounters.Instance;
  266. #if (WINDOWS_PHONE)
  267. private static readonly Dictionary<int, ThreadEntry> _threadEntries = new Dictionary<int, ThreadEntry>();
  268. #elif (_WINDOWS_CE)
  269. private static LocalDataStoreSlot _threadEntrySlot = Thread.AllocateDataSlot();
  270. #else
  271. [ThreadStatic]
  272. private static ThreadEntry _threadEntry;
  273. #endif
  274. /// <summary>
  275. /// An event to call after a thread is created, but before
  276. /// it's first use.
  277. /// </summary>
  278. private event ThreadInitializationHandler _onThreadInitialization;
  279. /// <summary>
  280. /// An event to call when a thread is about to exit, after
  281. /// it is no longer belong to the pool.
  282. /// </summary>
  283. private event ThreadTerminationHandler _onThreadTermination;
  284. #endregion
  285. #region Per thread properties
  286. /// <summary>
  287. /// A reference to the current work item a thread from the thread pool
  288. /// is executing.
  289. /// </summary>
  290. internal static ThreadEntry CurrentThreadEntry
  291. {
  292. #if (WINDOWS_PHONE)
  293. get
  294. {
  295. lock(_threadEntries)
  296. {
  297. ThreadEntry threadEntry;
  298. if (_threadEntries.TryGetValue(Thread.CurrentThread.ManagedThreadId, out threadEntry))
  299. {
  300. return threadEntry;
  301. }
  302. }
  303. return null;
  304. }
  305. set
  306. {
  307. lock(_threadEntries)
  308. {
  309. _threadEntries[Thread.CurrentThread.ManagedThreadId] = value;
  310. }
  311. }
  312. #elif (_WINDOWS_CE)
  313. get
  314. {
  315. //Thread.CurrentThread.ManagedThreadId
  316. return Thread.GetData(_threadEntrySlot) as ThreadEntry;
  317. }
  318. set
  319. {
  320. Thread.SetData(_threadEntrySlot, value);
  321. }
  322. #else
  323. get
  324. {
  325. return _threadEntry;
  326. }
  327. set
  328. {
  329. _threadEntry = value;
  330. }
  331. #endif
  332. }
  333. #endregion
  334. #region Construction and Finalization
  335. /// <summary>
  336. /// Constructor
  337. /// </summary>
  338. public SmartThreadPool()
  339. {
  340. _stpStartInfo = new STPStartInfo();
  341. Initialize();
  342. }
  343. /// <summary>
  344. /// Constructor
  345. /// </summary>
  346. /// <param name="idleTimeout">Idle timeout in milliseconds</param>
  347. public SmartThreadPool(int idleTimeout)
  348. {
  349. _stpStartInfo = new STPStartInfo
  350. {
  351. IdleTimeout = idleTimeout,
  352. };
  353. Initialize();
  354. }
  355. /// <summary>
  356. /// Constructor
  357. /// </summary>
  358. /// <param name="idleTimeout">Idle timeout in milliseconds</param>
  359. /// <param name="maxWorkerThreads">Upper limit of threads in the pool</param>
  360. public SmartThreadPool(
  361. int idleTimeout,
  362. int maxWorkerThreads)
  363. {
  364. _stpStartInfo = new STPStartInfo
  365. {
  366. IdleTimeout = idleTimeout,
  367. MaxWorkerThreads = maxWorkerThreads,
  368. };
  369. Initialize();
  370. }
  371. /// <summary>
  372. /// Constructor
  373. /// </summary>
  374. /// <param name="idleTimeout">Idle timeout in milliseconds</param>
  375. /// <param name="maxWorkerThreads">Upper limit of threads in the pool</param>
  376. /// <param name="minWorkerThreads">Lower limit of threads in the pool</param>
  377. public SmartThreadPool(
  378. int idleTimeout,
  379. int maxWorkerThreads,
  380. int minWorkerThreads)
  381. {
  382. _stpStartInfo = new STPStartInfo
  383. {
  384. IdleTimeout = idleTimeout,
  385. MaxWorkerThreads = maxWorkerThreads,
  386. MinWorkerThreads = minWorkerThreads,
  387. };
  388. Initialize();
  389. }
  390. /// <summary>
  391. /// Constructor
  392. /// </summary>
  393. /// <param name="stpStartInfo">A SmartThreadPool configuration that overrides the default behavior</param>
  394. public SmartThreadPool(STPStartInfo stpStartInfo)
  395. {
  396. _stpStartInfo = new STPStartInfo(stpStartInfo);
  397. Initialize();
  398. }
  399. private void Initialize()
  400. {
  401. Name = _stpStartInfo.ThreadPoolName;
  402. ValidateSTPStartInfo();
  403. // _stpStartInfoRW stores a read/write copy of the STPStartInfo.
  404. // Actually only MaxWorkerThreads and MinWorkerThreads are overwritten
  405. _isSuspended = _stpStartInfo.StartSuspended;
  406. #if (_WINDOWS_CE) || (_SILVERLIGHT) || (_MONO) || (WINDOWS_PHONE)
  407. if (null != _stpStartInfo.PerformanceCounterInstanceName)
  408. {
  409. throw new NotSupportedException("Performance counters are not implemented for Compact Framework/Silverlight/Mono, instead use StpStartInfo.EnableLocalPerformanceCounters");
  410. }
  411. #else
  412. if (null != _stpStartInfo.PerformanceCounterInstanceName)
  413. {
  414. try
  415. {
  416. _windowsPCs = new STPInstancePerformanceCounters(_stpStartInfo.PerformanceCounterInstanceName);
  417. }
  418. catch (Exception e)
  419. {
  420. Debug.WriteLine("Unable to create Performance Counters: " + e);
  421. _windowsPCs = NullSTPInstancePerformanceCounters.Instance;
  422. }
  423. }
  424. #endif
  425. if (_stpStartInfo.EnableLocalPerformanceCounters)
  426. {
  427. _localPCs = new LocalSTPInstancePerformanceCounters();
  428. }
  429. // If the STP is not started suspended then start the threads.
  430. if (!_isSuspended)
  431. {
  432. StartOptimalNumberOfThreads();
  433. }
  434. }
  435. private void StartOptimalNumberOfThreads()
  436. {
  437. int threadsCount = Math.Max(_workItemsQueue.Count, _stpStartInfo.MinWorkerThreads);
  438. threadsCount = Math.Min(threadsCount, _stpStartInfo.MaxWorkerThreads);
  439. threadsCount -= _workerThreads.Count;
  440. if (threadsCount > 0)
  441. {
  442. StartThreads(threadsCount);
  443. }
  444. }
  445. private void ValidateSTPStartInfo()
  446. {
  447. if (_stpStartInfo.MinWorkerThreads < 0)
  448. {
  449. throw new ArgumentOutOfRangeException(
  450. "MinWorkerThreads", "MinWorkerThreads cannot be negative");
  451. }
  452. if (_stpStartInfo.MaxWorkerThreads <= 0)
  453. {
  454. throw new ArgumentOutOfRangeException(
  455. "MaxWorkerThreads", "MaxWorkerThreads must be greater than zero");
  456. }
  457. if (_stpStartInfo.MinWorkerThreads > _stpStartInfo.MaxWorkerThreads)
  458. {
  459. throw new ArgumentOutOfRangeException(
  460. "MinWorkerThreads, maxWorkerThreads",
  461. "MaxWorkerThreads must be greater or equal to MinWorkerThreads");
  462. }
  463. }
  464. private static void ValidateCallback(Delegate callback)
  465. {
  466. if(callback.GetInvocationList().Length > 1)
  467. {
  468. throw new NotSupportedException("SmartThreadPool doesn't support delegates chains");
  469. }
  470. }
  471. #endregion
  472. #region Thread Processing
  473. /// <summary>
  474. /// Waits on the queue for a work item, shutdown, or timeout.
  475. /// </summary>
  476. /// <returns>
  477. /// Returns the WaitingCallback or null in case of timeout or shutdown.
  478. /// </returns>
  479. private WorkItem Dequeue()
  480. {
  481. WorkItem workItem =
  482. _workItemsQueue.DequeueWorkItem(_stpStartInfo.IdleTimeout, _shuttingDownEvent);
  483. return workItem;
  484. }
  485. /// <summary>
  486. /// Put a new work item in the queue
  487. /// </summary>
  488. /// <param name="workItem">A work item to queue</param>
  489. internal override void Enqueue(WorkItem workItem)
  490. {
  491. // Make sure the workItem is not null
  492. Debug.Assert(null != workItem);
  493. IncrementWorkItemsCount();
  494. workItem.CanceledSmartThreadPool = _canceledSmartThreadPool;
  495. _workItemsQueue.EnqueueWorkItem(workItem);
  496. workItem.WorkItemIsQueued();
  497. // If all the threads are busy then try to create a new one
  498. if (_currentWorkItemsCount > _workerThreads.Count)
  499. {
  500. StartThreads(1);
  501. }
  502. }
  503. private void IncrementWorkItemsCount()
  504. {
  505. _windowsPCs.SampleWorkItems(_workItemsQueue.Count, _workItemsProcessed);
  506. _localPCs.SampleWorkItems(_workItemsQueue.Count, _workItemsProcessed);
  507. int count = Interlocked.Increment(ref _currentWorkItemsCount);
  508. //Trace.WriteLine("WorkItemsCount = " + _currentWorkItemsCount.ToString());
  509. if (count == 1)
  510. {
  511. IsIdle = false;
  512. _isIdleWaitHandle.Reset();
  513. }
  514. }
  515. private void DecrementWorkItemsCount()
  516. {
  517. int count = Interlocked.Decrement(ref _currentWorkItemsCount);
  518. //Trace.WriteLine("WorkItemsCount = " + _currentWorkItemsCount.ToString());
  519. if (count == 0)
  520. {
  521. IsIdle = true;
  522. _isIdleWaitHandle.Set();
  523. }
  524. Interlocked.Increment(ref _workItemsProcessed);
  525. if (!_shutdown)
  526. {
  527. // The counter counts even if the work item was cancelled
  528. _windowsPCs.SampleWorkItems(_workItemsQueue.Count, _workItemsProcessed);
  529. _localPCs.SampleWorkItems(_workItemsQueue.Count, _workItemsProcessed);
  530. }
  531. }
  532. internal void RegisterWorkItemsGroup(IWorkItemsGroup workItemsGroup)
  533. {
  534. _workItemsGroups[workItemsGroup] = workItemsGroup;
  535. }
  536. internal void UnregisterWorkItemsGroup(IWorkItemsGroup workItemsGroup)
  537. {
  538. if (_workItemsGroups.Contains(workItemsGroup))
  539. {
  540. _workItemsGroups.Remove(workItemsGroup);
  541. }
  542. }
  543. /// <summary>
  544. /// Inform that the current thread is about to quit or quiting.
  545. /// The same thread may call this method more than once.
  546. /// </summary>
  547. private void InformCompleted()
  548. {
  549. // There is no need to lock the two methods together
  550. // since only the current thread removes itself
  551. // and the _workerThreads is a synchronized dictionary
  552. if (_workerThreads.Contains(Thread.CurrentThread))
  553. {
  554. _workerThreads.Remove(Thread.CurrentThread);
  555. _windowsPCs.SampleThreads(_workerThreads.Count, _inUseWorkerThreads);
  556. _localPCs.SampleThreads(_workerThreads.Count, _inUseWorkerThreads);
  557. }
  558. }
  559. /// <summary>
  560. /// Starts new threads
  561. /// </summary>
  562. /// <param name="threadsCount">The number of threads to start</param>
  563. private void StartThreads(int threadsCount)
  564. {
  565. if (_isSuspended)
  566. {
  567. return;
  568. }
  569. lock(_workerThreads.SyncRoot)
  570. {
  571. // Don't start threads on shut down
  572. if (_shutdown)
  573. {
  574. return;
  575. }
  576. for(int i = 0; i < threadsCount; ++i)
  577. {
  578. // Don't create more threads then the upper limit
  579. if (_workerThreads.Count >= _stpStartInfo.MaxWorkerThreads)
  580. {
  581. return;
  582. }
  583. // Create a new thread
  584. #if (_SILVERLIGHT) || (WINDOWS_PHONE)
  585. Thread workerThread = new Thread(ProcessQueuedItems);
  586. #else
  587. Thread workerThread =
  588. _stpStartInfo.MaxStackSize.HasValue
  589. ? new Thread(ProcessQueuedItems, _stpStartInfo.MaxStackSize.Value)
  590. : new Thread(ProcessQueuedItems);
  591. #endif
  592. // Configure the new thread and start it
  593. workerThread.IsBackground = _stpStartInfo.AreThreadsBackground;
  594. #if !(_SILVERLIGHT) && !(_WINDOWS_CE) && !(WINDOWS_PHONE)
  595. if (_stpStartInfo.ApartmentState != ApartmentState.Unknown)
  596. {
  597. workerThread.SetApartmentState(_stpStartInfo.ApartmentState);
  598. }
  599. #endif
  600. #if !(_SILVERLIGHT) && !(WINDOWS_PHONE)
  601. workerThread.Priority = _stpStartInfo.ThreadPriority;
  602. #endif
  603. workerThread.Name = string.Format("STP:{0}:{1}", Name, _threadCounter);
  604. workerThread.Start();
  605. ++_threadCounter;
  606. // Add it to the dictionary and update its creation time.
  607. _workerThreads[workerThread] = new ThreadEntry(this);
  608. _windowsPCs.SampleThreads(_workerThreads.Count, _inUseWorkerThreads);
  609. _localPCs.SampleThreads(_workerThreads.Count, _inUseWorkerThreads);
  610. }
  611. }
  612. }
  613. /// <summary>
  614. /// A worker thread method that processes work items from the work items queue.
  615. /// </summary>
  616. private void ProcessQueuedItems()
  617. {
  618. // Keep the entry of the dictionary as thread's variable to avoid the synchronization locks
  619. // of the dictionary.
  620. CurrentThreadEntry = _workerThreads[Thread.CurrentThread];
  621. FireOnThreadInitialization();
  622. try
  623. {
  624. bool bInUseWorkerThreadsWasIncremented = false;
  625. // Process until shutdown.
  626. while(!_shutdown)
  627. {
  628. // Update the last time this thread was seen alive.
  629. // It's good for debugging.
  630. CurrentThreadEntry.IAmAlive();
  631. // The following block handles the when the MaxWorkerThreads has been
  632. // incremented by the user at run-time.
  633. // Double lock for quit.
  634. if (_workerThreads.Count > _stpStartInfo.MaxWorkerThreads)
  635. {
  636. lock (_workerThreads.SyncRoot)
  637. {
  638. if (_workerThreads.Count > _stpStartInfo.MaxWorkerThreads)
  639. {
  640. // Inform that the thread is quiting and then quit.
  641. // This method must be called within this lock or else
  642. // more threads will quit and the thread pool will go
  643. // below the lower limit.
  644. InformCompleted();
  645. break;
  646. }
  647. }
  648. }
  649. // Wait for a work item, shutdown, or timeout
  650. WorkItem workItem = Dequeue();
  651. // Update the last time this thread was seen alive.
  652. // It's good for debugging.
  653. CurrentThreadEntry.IAmAlive();
  654. // On timeout or shut down.
  655. if (null == workItem)
  656. {
  657. // Double lock for quit.
  658. if (_workerThreads.Count > _stpStartInfo.MinWorkerThreads)
  659. {
  660. lock(_workerThreads.SyncRoot)
  661. {
  662. if (_workerThreads.Count > _stpStartInfo.MinWorkerThreads)
  663. {
  664. // Inform that the thread is quiting and then quit.
  665. // This method must be called within this lock or else
  666. // more threads will quit and the thread pool will go
  667. // below the lower limit.
  668. InformCompleted();
  669. break;
  670. }
  671. }
  672. }
  673. }
  674. // If we didn't quit then skip to the next iteration.
  675. if (null == workItem)
  676. {
  677. continue;
  678. }
  679. try
  680. {
  681. // Initialize the value to false
  682. bInUseWorkerThreadsWasIncremented = false;
  683. // Set the Current Work Item of the thread.
  684. // Store the Current Work Item before the workItem.StartingWorkItem() is called,
  685. // so WorkItem.Cancel can work when the work item is between InQueue and InProgress
  686. // states.
  687. // If the work item has been cancelled BEFORE the workItem.StartingWorkItem()
  688. // (work item is in InQueue state) then workItem.StartingWorkItem() will return false.
  689. // If the work item has been cancelled AFTER the workItem.StartingWorkItem() then
  690. // (work item is in InProgress state) then the thread will be aborted
  691. CurrentThreadEntry.CurrentWorkItem = workItem;
  692. // Change the state of the work item to 'in progress' if possible.
  693. // We do it here so if the work item has been canceled we won't
  694. // increment the _inUseWorkerThreads.
  695. // The cancel mechanism doesn't delete items from the queue,
  696. // it marks the work item as canceled, and when the work item
  697. // is dequeued, we just skip it.
  698. // If the post execute of work item is set to always or to
  699. // call when the work item is canceled then the StartingWorkItem()
  700. // will return true, so the post execute can run.
  701. if (!workItem.StartingWorkItem())
  702. {
  703. continue;
  704. }
  705. // Execute the callback. Make sure to accurately
  706. // record how many callbacks are currently executing.
  707. int inUseWorkerThreads = Interlocked.Increment(ref _inUseWorkerThreads);
  708. _windowsPCs.SampleThreads(_workerThreads.Count, inUseWorkerThreads);
  709. _localPCs.SampleThreads(_workerThreads.Count, inUseWorkerThreads);
  710. // Mark that the _inUseWorkerThreads incremented, so in the finally{}
  711. // statement we will decrement it correctly.
  712. bInUseWorkerThreadsWasIncremented = true;
  713. workItem.FireWorkItemStarted();
  714. ExecuteWorkItem(workItem);
  715. }
  716. catch(Exception ex)
  717. {
  718. ex.GetHashCode();
  719. // Do nothing
  720. }
  721. finally
  722. {
  723. workItem.DisposeOfState();
  724. // Set the CurrentWorkItem to null, since we
  725. // no longer run user's code.
  726. CurrentThreadEntry.CurrentWorkItem = null;
  727. // Decrement the _inUseWorkerThreads only if we had
  728. // incremented it. Note the cancelled work items don't
  729. // increment _inUseWorkerThreads.
  730. if (bInUseWorkerThreadsWasIncremented)
  731. {
  732. int inUseWorkerThreads = Interlocked.Decrement(ref _inUseWorkerThreads);
  733. _windowsPCs.SampleThreads(_workerThreads.Count, inUseWorkerThreads);
  734. _localPCs.SampleThreads(_workerThreads.Count, inUseWorkerThreads);
  735. }
  736. // Notify that the work item has been completed.
  737. // WorkItemsGroup may enqueue their next work item.
  738. workItem.FireWorkItemCompleted();
  739. // Decrement the number of work items here so the idle
  740. // ManualResetEvent won't fluctuate.
  741. DecrementWorkItemsCount();
  742. }
  743. }
  744. }
  745. catch(ThreadAbortException tae)
  746. {
  747. tae.GetHashCode();
  748. // Handle the abort exception gracfully.
  749. #if !(_WINDOWS_CE) && !(_SILVERLIGHT) && !(WINDOWS_PHONE)
  750. Thread.ResetAbort();
  751. #endif
  752. }
  753. catch(Exception e)
  754. {
  755. Debug.Assert(null != e);
  756. }
  757. finally
  758. {
  759. InformCompleted();
  760. FireOnThreadTermination();
  761. }
  762. }
  763. private void ExecuteWorkItem(WorkItem workItem)
  764. {
  765. _windowsPCs.SampleWorkItemsWaitTime(workItem.WaitingTime);
  766. _localPCs.SampleWorkItemsWaitTime(workItem.WaitingTime);
  767. try
  768. {
  769. workItem.Execute();
  770. }
  771. finally
  772. {
  773. _windowsPCs.SampleWorkItemsProcessTime(workItem.ProcessTime);
  774. _localPCs.SampleWorkItemsProcessTime(workItem.ProcessTime);
  775. }
  776. }
  777. #endregion
  778. #region Public Methods
  779. private void ValidateWaitForIdle()
  780. {
  781. if (null != CurrentThreadEntry && CurrentThreadEntry.AssociatedSmartThreadPool == this)
  782. {
  783. throw new NotSupportedException(
  784. "WaitForIdle cannot be called from a thread on its SmartThreadPool, it causes a deadlock");
  785. }
  786. }
  787. internal static void ValidateWorkItemsGroupWaitForIdle(IWorkItemsGroup workItemsGroup)
  788. {
  789. if (null == CurrentThreadEntry)
  790. {
  791. return;
  792. }
  793. WorkItem workItem = CurrentThreadEntry.CurrentWorkItem;
  794. ValidateWorkItemsGroupWaitForIdleImpl(workItemsGroup, workItem);
  795. if ((null != workItemsGroup) &&
  796. (null != workItem) &&
  797. CurrentThreadEntry.CurrentWorkItem.WasQueuedBy(workItemsGroup))
  798. {
  799. throw new NotSupportedException("WaitForIdle cannot be called from a thread on its SmartThreadPool, it causes a deadlock");
  800. }
  801. }
  802. [MethodImpl(MethodImplOptions.NoInlining)]
  803. private static void ValidateWorkItemsGroupWaitForIdleImpl(IWorkItemsGroup workItemsGroup, WorkItem workItem)
  804. {
  805. if ((null != workItemsGroup) &&
  806. (null != workItem) &&
  807. workItem.WasQueuedBy(workItemsGroup))
  808. {
  809. throw new NotSupportedException("WaitForIdle cannot be called from a thread on its SmartThreadPool, it causes a deadlock");
  810. }
  811. }
  812. /// <summary>
  813. /// Force the SmartThreadPool to shutdown
  814. /// </summary>
  815. public void Shutdown()
  816. {
  817. Shutdown(true, 0);
  818. }
  819. /// <summary>
  820. /// Force the SmartThreadPool to shutdown with timeout
  821. /// </summary>
  822. public void Shutdown(bool forceAbort, TimeSpan timeout)
  823. {
  824. Shutdown(forceAbort, (int)timeout.TotalMilliseconds);
  825. }
  826. /// <summary>
  827. /// Empties the queue of work items and abort the threads in the pool.
  828. /// </summary>
  829. public void Shutdown(bool forceAbort, int millisecondsTimeout)
  830. {
  831. ValidateNotDisposed();
  832. ISTPInstancePerformanceCounters pcs = _windowsPCs;
  833. if (NullSTPInstancePerformanceCounters.Instance != _windowsPCs)
  834. {
  835. // Set the _pcs to "null" to stop updating the performance
  836. // counters
  837. _windowsPCs = NullSTPInstancePerformanceCounters.Instance;
  838. pcs.Dispose();
  839. }
  840. Thread [] threads;
  841. lock(_workerThreads.SyncRoot)
  842. {
  843. // Shutdown the work items queue
  844. _workItemsQueue.Dispose();
  845. // Signal the threads to exit
  846. _shutdown = true;
  847. _shuttingDownEvent.Set();
  848. // Make a copy of the threads' references in the pool
  849. threads = new Thread [_workerThreads.Count];
  850. _workerThreads.Keys.CopyTo(threads, 0);
  851. }
  852. int millisecondsLeft = millisecondsTimeout;
  853. Stopwatch stopwatch = Stopwatch.StartNew();
  854. //DateTime start = DateTime.UtcNow;
  855. bool waitInfinitely = (Timeout.Infinite == millisecondsTimeout);
  856. bool timeout = false;
  857. // Each iteration we update the time left for the timeout.
  858. foreach(Thread thread in threads)
  859. {
  860. // Join don't work with negative numbers
  861. if (!waitInfinitely && (millisecondsLeft < 0))
  862. {
  863. timeout = true;
  864. break;
  865. }
  866. // Wait for the thread to terminate
  867. bool success = thread.Join(millisecondsLeft);
  868. if(!success)
  869. {
  870. timeout = true;
  871. break;
  872. }
  873. if(!waitInfinitely)
  874. {
  875. // Update the time left to wait
  876. //TimeSpan ts = DateTime.UtcNow - start;
  877. millisecondsLeft = millisecondsTimeout - (int)stopwatch.ElapsedMilliseconds;
  878. }
  879. }
  880. if (timeout && forceAbort)
  881. {
  882. // Abort the threads in the pool
  883. foreach(Thread thread in threads)
  884. {
  885. if ((thread != null)
  886. #if !(_WINDOWS_CE)
  887. && thread.IsAlive
  888. #endif
  889. )
  890. {
  891. try
  892. {
  893. thread.Abort(); // Shutdown
  894. }
  895. catch(SecurityException e)
  896. {
  897. e.GetHashCode();
  898. }
  899. catch(ThreadStateException ex)
  900. {
  901. ex.GetHashCode();
  902. // In case the thread has been terminated
  903. // after the check if it is alive.
  904. }
  905. }
  906. }
  907. }
  908. }
  909. /// <summary>
  910. /// Wait for all work items to complete
  911. /// </summary>
  912. /// <param name="waitableResults">Array of work item result objects</param>
  913. /// <returns>
  914. /// true when every work item in workItemResults has completed; otherwise false.
  915. /// </returns>
  916. public static bool WaitAll(
  917. IWaitableResult [] waitableResults)
  918. {
  919. return WaitAll(waitableResults, Timeout.Infinite, true);
  920. }
  921. /// <summary>
  922. /// Wait for all work items to complete
  923. /// </summary>
  924. /// <param name="waitableResults">Array of work item result objects</param>
  925. /// <param name="timeout">The number of milliseconds to wait, or a TimeSpan that represents -1 milliseconds to wait indefinitely. </param>
  926. /// <param name="exitContext">
  927. /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
  928. /// </param>
  929. /// <returns>
  930. /// true when every work item in workItemResults has completed; otherwise false.
  931. /// </returns>
  932. public static bool WaitAll(
  933. IWaitableResult [] waitableResults,
  934. TimeSpan timeout,
  935. bool exitContext)
  936. {
  937. return WaitAll(waitableResults, (int)timeout.TotalMilliseconds, exitContext);
  938. }
  939. /// <summary>
  940. /// Wait for all work items to complete
  941. /// </summary>
  942. /// <param name="waitableResults">Array of work item result objects</param>
  943. /// <param name="timeout">The number of milliseconds to wait, or a TimeSpan that represents -1 milliseconds to wait indefinitely. </param>
  944. /// <param name="exitContext">
  945. /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
  946. /// </param>
  947. /// <param name="cancelWaitHandle">A cancel wait handle to interrupt the wait if needed</param>
  948. /// <returns>
  949. /// true when every work item in workItemResults has completed; otherwise false.
  950. /// </returns>
  951. public static bool WaitAll(
  952. IWaitableResult[] waitableResults,
  953. TimeSpan timeout,
  954. bool exitContext,
  955. WaitHandle cancelWaitHandle)
  956. {
  957. return WaitAll(waitableResults, (int)timeout.TotalMilliseconds, exitContext, cancelWaitHandle);
  958. }
  959. /// <summary>
  960. /// Wait for all work items to complete
  961. /// </summary>
  962. /// <param name="waitableResults">Array of work item result objects</param>
  963. /// <param name="millisecondsTimeout">The number of milliseconds to wait, or Timeout.Infinite (-1) to wait indefinitely.</param>
  964. /// <param name="exitContext">
  965. /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
  966. /// </param>
  967. /// <returns>
  968. /// true when every work item in workItemResults has completed; otherwise false.
  969. /// </returns>
  970. public static bool WaitAll(
  971. IWaitableResult [] waitableResults,
  972. int millisecondsTimeout,
  973. bool exitContext)
  974. {
  975. return WorkItem.WaitAll(waitableResults, millisecondsTimeout, exitContext, null);
  976. }
  977. /// <summary>
  978. /// Wait for all work items to complete
  979. /// </summary>
  980. /// <param name="waitableResults">Array of work item result objects</param>
  981. /// <param name="millisecondsTimeout">The number of milliseconds to wait, or Timeout.Infinite (-1) to wait indefinitely.</param>
  982. /// <param name="exitContext">
  983. /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
  984. /// </param>
  985. /// <param name="cancelWaitHandle">A cancel wait handle to interrupt the wait if needed</param>
  986. /// <returns>
  987. /// true when every work item in workItemResults has completed; otherwise false.
  988. /// </returns>
  989. public static bool WaitAll(
  990. IWaitableResult[] waitableResults,
  991. int millisecondsTimeout,
  992. bool exitContext,
  993. WaitHandle cancelWaitHandle)
  994. {
  995. return WorkItem.WaitAll(waitableResults, millisecondsTimeout, exitContext, cancelWaitHandle);
  996. }
  997. /// <summary>
  998. /// Waits for any of the work items in the specified array to complete, cancel, or timeout
  999. /// </summary>
  1000. /// <param name="waitableResults">Array of work item result objects</param>
  1001. /// <returns>
  1002. /// The array index of the work item result that satisfied the wait, or WaitTimeout if any of the work items has been canceled.
  1003. /// </returns>
  1004. public static int WaitAny(
  1005. IWaitableResult [] waitableResults)
  1006. {
  1007. return WaitAny(waitableResults, Timeout.Infinite, true);
  1008. }
  1009. /// <summary>
  1010. /// Waits for any of the work items in the specified array to complete, cancel, or timeout
  1011. /// </summary>
  1012. /// <param name="waitableResults">Array of work item result objects</param>
  1013. /// <param name="timeout">The number of milliseconds to wait, or a TimeSpan that represents -1 milliseconds to wait indefinitely. </param>
  1014. /// <param name="exitContext">
  1015. /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
  1016. /// </param>
  1017. /// <returns>
  1018. /// 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.
  1019. /// </returns>
  1020. public static int WaitAny(
  1021. IWaitableResult[] waitableResults,
  1022. TimeSpan timeout,
  1023. bool exitContext)
  1024. {
  1025. return WaitAny(waitableResults, (int)timeout.TotalMilliseconds, exitContext);
  1026. }
  1027. /// <summary>
  1028. /// Waits for any of the work items in the specified array to complete, cancel, or timeout
  1029. /// </summary>
  1030. /// <param name="waitableResults">Array of work item result objects</param>
  1031. /// <param name="timeout">The number of milliseconds to wait, or a TimeSpan that represents -1 milliseconds to wait indefinitely. </param>
  1032. /// <param name="exitContext">
  1033. /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
  1034. /// </param>
  1035. /// <param name="cancelWaitHandle">A cancel wait handle to interrupt the wait if needed</param>
  1036. /// <returns>
  1037. /// 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.
  1038. /// </returns>
  1039. public static int WaitAny(
  1040. IWaitableResult [] waitableResults,
  1041. TimeSpan timeout,
  1042. bool exitContext,
  1043. WaitHandle cancelWaitHandle)
  1044. {
  1045. return WaitAny(waitableResults, (int)timeout.TotalMilliseconds, exitContext, cancelWaitHandle);
  1046. }
  1047. /// <summary>
  1048. /// Waits for any of the work items in the specified array to complete, cancel, or timeout
  1049. /// </summary>
  1050. /// <param name="waitableResults">Array of work item result objects</param>
  1051. /// <param name="millisecondsTimeout">The number of milliseconds to wait, or Timeout.Infinite (-1) to wait indefinitely.</param>
  1052. /// <param name="exitContext">
  1053. /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
  1054. /// </param>
  1055. /// <returns>
  1056. /// 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.
  1057. /// </returns>
  1058. public static int WaitAny(
  1059. IWaitableResult [] waitableResults,
  1060. int millisecondsTimeout,
  1061. bool exitContext)
  1062. {
  1063. return WorkItem.WaitAny(waitableResults, millisecondsTimeout, exitContext, null);
  1064. }
  1065. /// <summary>
  1066. /// Waits for any of the work items in the specified array to complete, cancel, or timeout
  1067. /// </summary>
  1068. /// <param name="waitableResults">Array of work item result objects</param>
  1069. /// <param name="millisecondsTimeout">The number of milliseconds to wait, or Timeout.Infinite (-1) to wait indefinitely.</param>
  1070. /// <param name="exitContext">
  1071. /// true to exit the synchronization domain for the context before the wait (if in a synchronized context), and reacquire it; otherwise, false.
  1072. /// </param>
  1073. /// <param name="cancelWaitHandle">A cancel wait handle to interrupt the wait if needed</param>
  1074. /// <returns>
  1075. /// 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.
  1076. /// </returns>
  1077. public static int WaitAny(
  1078. IWaitableResult [] waitableResults,
  1079. int millisecondsTimeout,
  1080. bool exitContext,
  1081. WaitHandle cancelWaitHandle)
  1082. {
  1083. return WorkItem.WaitAny(waitableResults, millisecondsTimeout, exitContext, cancelWaitHandle);
  1084. }
  1085. /// <summary>
  1086. /// Creates a new WorkItemsGroup.
  1087. /// </summary>
  1088. /// <param name="concurrency">The number of work items that can be run concurrently</param>
  1089. /// <returns>A reference to the WorkItemsGroup</returns>
  1090. public IWorkItemsGroup CreateWorkItemsGroup(int concurrency)
  1091. {
  1092. IWorkItemsGroup workItemsGroup = new WorkItemsGroup(this, concurrency, _stpStartInfo);
  1093. return workItemsGroup;
  1094. }
  1095. /// <summary>
  1096. /// Creates a new WorkItemsGroup.
  1097. /// </summary>
  1098. /// <param name="concurrency">The number of work items that can be run concurrently</param>
  1099. /// <param name="wigStartInfo">A WorkItemsGroup configuration that overrides the default behavior</param>
  1100. /// <returns>A reference to the WorkItemsGroup</returns>
  1101. public IWorkItemsGroup CreateWorkItemsGroup(int concurrency, WIGStartInfo wigStartInfo)
  1102. {
  1103. IWorkItemsGroup workItemsGroup = new WorkItemsGroup(this, concurrency, wigStartInfo);
  1104. return workItemsGroup;
  1105. }
  1106. #region Fire Thread's Events
  1107. private void FireOnThreadInitialization()
  1108. {
  1109. if (null != _onThreadInitialization)
  1110. {
  1111. foreach (ThreadInitializationHandler tih in _onThreadInitialization.GetInvocationList())
  1112. {
  1113. try
  1114. {
  1115. tih();
  1116. }
  1117. catch (Exception e)
  1118. {
  1119. e.GetHashCode();
  1120. Debug.Assert(false);
  1121. throw;
  1122. }
  1123. }
  1124. }
  1125. }
  1126. private void FireOnThreadTermination()
  1127. {
  1128. if (null != _onThreadTermination)
  1129. {
  1130. foreach (ThreadTerminationHandler tth in _onThreadTermination.GetInvocationList())
  1131. {
  1132. try
  1133. {
  1134. tth();
  1135. }
  1136. catch (Exception e)
  1137. {
  1138. e.GetHashCode();
  1139. Debug.Assert(false);
  1140. throw;
  1141. }
  1142. }
  1143. }
  1144. }
  1145. #endregion
  1146. /// <summary>
  1147. /// This event is fired when a thread is created.
  1148. /// Use it to initialize a thread before the work items use it.
  1149. /// </summary>
  1150. public event ThreadInitializationHandler OnThreadInitialization
  1151. {
  1152. add { _onThreadInitialization += value; }
  1153. remove { _onThreadInitialization -= value; }
  1154. }
  1155. /// <summary>
  1156. /// This event is fired when a thread is terminating.
  1157. /// Use it for cleanup.
  1158. /// </summary>
  1159. public event ThreadTerminationHandler OnThreadTermination
  1160. {
  1161. add { _onThreadTermination += value; }
  1162. remove { _onThreadTermination -= value; }
  1163. }
  1164. internal void CancelAbortWorkItemsGroup(WorkItemsGroup wig)
  1165. {
  1166. foreach (ThreadEntry threadEntry in _workerThreads.Values)
  1167. {
  1168. WorkItem workItem = threadEntry.CurrentWorkItem;
  1169. if (null != workItem &&
  1170. workItem.WasQueuedBy(wig) &&
  1171. !workItem.IsCanceled)
  1172. {
  1173. threadEntry.CurrentWorkItem.GetWorkItemResult().Cancel(true);
  1174. }
  1175. }
  1176. }
  1177. #endregion
  1178. #region Properties
  1179. /// <summary>
  1180. /// Get/Set the lower limit of threads in the pool.
  1181. /// </summary>
  1182. public int MinThreads
  1183. {
  1184. get
  1185. {
  1186. ValidateNotDisposed();
  1187. return _stpStartInfo.MinWorkerThreads;
  1188. }
  1189. set
  1190. {
  1191. Debug.Assert(value >= 0);
  1192. Debug.Assert(value <= _stpStartInfo.MaxWorkerThreads);
  1193. if (_stpStartInfo.MaxWorkerThreads < value)
  1194. {
  1195. _stpStartInfo.MaxWorkerThreads = value;
  1196. }
  1197. _stpStartInfo.MinWorkerThreads = value;
  1198. StartOptimalNumberOfThreads();
  1199. }
  1200. }
  1201. /// <summary>
  1202. /// Get/Set the upper limit of threads in the pool.
  1203. /// </summary>
  1204. public int MaxThreads
  1205. {
  1206. get
  1207. {
  1208. ValidateNotDisposed();
  1209. return _stpStartInfo.MaxWorkerThreads;
  1210. }
  1211. set
  1212. {
  1213. Debug.Assert(value > 0);
  1214. Debug.Assert(value >= _stpStartInfo.MinWorkerThreads);
  1215. if (_stpStartInfo.MinWorkerThreads > value)
  1216. {
  1217. _stpStartInfo.MinWorkerThreads = value;
  1218. }
  1219. _stpStartInfo.MaxWorkerThreads = value;
  1220. StartOptimalNumberOfThreads();
  1221. }
  1222. }
  1223. /// <summary>
  1224. /// Get the number of threads in the thread pool.
  1225. /// Should be between the lower and the upper limits.
  1226. /// </summary>
  1227. public int ActiveThreads
  1228. {
  1229. get
  1230. {
  1231. ValidateNotDisposed();
  1232. return _workerThreads.Count;
  1233. }
  1234. }
  1235. /// <summary>
  1236. /// Get the number of busy (not idle) threads in the thread pool.
  1237. /// </summary>
  1238. public int InUseThreads
  1239. {
  1240. get
  1241. {
  1242. ValidateNotDisposed();
  1243. return _inUseWorkerThreads;
  1244. }
  1245. }
  1246. /// <summary>
  1247. /// Returns true if the current running work item has been cancelled.
  1248. /// Must be used within the work item's callback method.
  1249. /// The work item should sample this value in order to know if it
  1250. /// needs to quit before its completion.
  1251. /// </summary>
  1252. public static bool IsWorkItemCanceled
  1253. {
  1254. get
  1255. {
  1256. return CurrentThreadEntry.CurrentWorkItem.IsCanceled;
  1257. }
  1258. }
  1259. /// <summary>
  1260. /// Checks if the work item has been cancelled, and if yes then abort the thread.
  1261. /// Can be used with Cancel and timeout
  1262. /// </summary>
  1263. public static void AbortOnWorkItemCancel()
  1264. {
  1265. if (IsWorkItemCanceled)
  1266. {
  1267. Thread.CurrentThread.Abort();
  1268. }
  1269. }
  1270. /// <summary>
  1271. /// Thread Pool start information (readonly)
  1272. /// </summary>
  1273. public STPStartInfo STPStartInfo
  1274. {
  1275. get
  1276. {
  1277. return _stpStartInfo.AsReadOnly();
  1278. }
  1279. }
  1280. public bool IsShuttingdown
  1281. {
  1282. get { return _shutdown; }
  1283. }
  1284. /// <summary>
  1285. /// Return the local calculated performance counters
  1286. /// Available only if STPStartInfo.EnableLocalPerformanceCounters is true.
  1287. /// </summary>
  1288. public ISTPPerformanceCountersReader PerformanceCountersReader
  1289. {
  1290. get { return (ISTPPerformanceCountersReader)_localPCs; }
  1291. }
  1292. #endregion
  1293. #region IDisposable Members
  1294. public void Dispose()
  1295. {
  1296. if (!_isDisposed)
  1297. {
  1298. if (!_shutdown)
  1299. {
  1300. Shutdown();
  1301. }
  1302. if (null != _shuttingDownEvent)
  1303. {
  1304. _shuttingDownEvent.Close();
  1305. _shuttingDownEvent = null;
  1306. }
  1307. _workerThreads.Clear();
  1308. if (null != _isIdleWaitHandle)
  1309. {
  1310. _isIdleWaitHandle.Close();
  1311. _isIdleWaitHandle = null;
  1312. }
  1313. if (_stpStartInfo.EnableLocalPerformanceCounters)
  1314. _localPCs.Dispose();
  1315. _isDisposed = true;
  1316. }
  1317. }
  1318. private void ValidateNotDisposed()
  1319. {
  1320. if(_isDisposed)
  1321. {
  1322. throw new ObjectDisposedException(GetType().ToString(), "The SmartThreadPool has been shutdown");
  1323. }
  1324. }
  1325. #endregion
  1326. #region WorkItemsGroupBase Overrides
  1327. /// <summary>
  1328. /// Get/Set the maximum number of work items that execute cocurrency on the thread pool
  1329. /// </summary>
  1330. public override int Concurrency
  1331. {
  1332. get { return MaxThreads; }
  1333. set { MaxThreads = value; }
  1334. }
  1335. /// <summary>
  1336. /// Get the number of work items in the queue.
  1337. /// </summary>
  1338. public override int WaitingCallbacks
  1339. {
  1340. get
  1341. {
  1342. ValidateNotDisposed();
  1343. return _workItemsQueue.Count;
  1344. }
  1345. }
  1346. /// <summary>
  1347. /// Get an array with all the state objects of the currently running items.
  1348. /// The array represents a snap shot and impact performance.
  1349. /// </summary>
  1350. public override object[] GetStates()
  1351. {
  1352. object[] states = _workItemsQueue.GetStates();
  1353. return states;
  1354. }
  1355. /// <summary>
  1356. /// WorkItemsGroup start information (readonly)
  1357. /// </summary>
  1358. public override WIGStartInfo WIGStartInfo
  1359. {
  1360. get { return _stpStartInfo.AsReadOnly(); }
  1361. }
  1362. /// <summary>
  1363. /// Start the thread pool if it was started suspended.
  1364. /// If it is already running, this method is ignored.
  1365. /// </summary>
  1366. public override void Start()
  1367. {
  1368. if (!_isSuspended)
  1369. {
  1370. return;
  1371. }
  1372. _isSuspended = false;
  1373. ICollection workItemsGroups = _workItemsGroups.Values;
  1374. foreach (WorkItemsGroup workItemsGroup in workItemsGroups)
  1375. {
  1376. workItemsGroup.OnSTPIsStarting();
  1377. }
  1378. StartOptimalNumberOfThreads();
  1379. }
  1380. /// <summary>
  1381. /// Cancel all work items using thread abortion
  1382. /// </summary>
  1383. /// <param name="abortExecution">True to stop work items by raising ThreadAbortException</param>
  1384. public override void Cancel(bool abortExecution)
  1385. {
  1386. _canceledSmartThreadPool.IsCanceled = true;
  1387. _canceledSmartThreadPool = new CanceledWorkItemsGroup();
  1388. ICollection workItemsGroups = _workItemsGroups.Values;
  1389. foreach (WorkItemsGroup workItemsGroup in workItemsGroups)
  1390. {
  1391. workItemsGroup.Cancel(abortExecution);
  1392. }
  1393. if (abortExecution)
  1394. {
  1395. foreach (ThreadEntry threadEntry in _workerThreads.Values)
  1396. {
  1397. WorkItem workItem = threadEntry.CurrentWorkItem;
  1398. if (null != workItem &&
  1399. threadEntry.AssociatedSmartThreadPool == this &&
  1400. !workItem.IsCanceled)
  1401. {
  1402. threadEntry.CurrentWorkItem.GetWorkItemResult().Cancel(true);
  1403. }
  1404. }
  1405. }
  1406. }
  1407. /// <summary>
  1408. /// Wait for the thread pool to be idle
  1409. /// </summary>
  1410. public override bool WaitForIdle(int millisecondsTimeout)
  1411. {
  1412. ValidateWaitForIdle();
  1413. return STPEventWaitHandle.WaitOne(_isIdleWaitHandle, millisecondsTimeout, false);
  1414. }
  1415. /// <summary>
  1416. /// This event is fired when all work items are completed.
  1417. /// (When IsIdle changes to true)
  1418. /// This event only work on WorkItemsGroup. On SmartThreadPool
  1419. /// it throws the NotImplementedException.
  1420. /// </summary>
  1421. public override event WorkItemsGroupIdleHandler OnIdle
  1422. {
  1423. add
  1424. {
  1425. throw new NotImplementedException("This event is not implemented in the SmartThreadPool class. Please create a WorkItemsGroup in order to use this feature.");
  1426. //_onIdle += value;
  1427. }
  1428. remove
  1429. {
  1430. throw new NotImplementedException("This event is not implemented in the SmartThreadPool class. Please create a WorkItemsGroup in order to use this feature.");
  1431. //_onIdle -= value;
  1432. }
  1433. }
  1434. internal override void PreQueueWorkItem()
  1435. {
  1436. ValidateNotDisposed();
  1437. }
  1438. #endregion
  1439. #region Join, Choice, Pipe, etc.
  1440. /// <summary>
  1441. /// Executes all actions in parallel.
  1442. /// Returns when they all finish.
  1443. /// </summary>
  1444. /// <param name="actions">Actions to execute</param>
  1445. public void Join(IEnumerable<Action> actions)
  1446. {
  1447. WIGStartInfo wigStartInfo = new WIGStartInfo { StartSuspended = true };
  1448. IWorkItemsGroup workItemsGroup = CreateWorkItemsGroup(int.MaxValue, wigStartInfo);
  1449. foreach (Action action in actions)
  1450. {
  1451. workItemsGroup.QueueWorkItem(action);
  1452. }
  1453. workItemsGroup.Start();
  1454. workItemsGroup.WaitForIdle();
  1455. }
  1456. /// <summary>
  1457. /// Executes all actions in parallel.
  1458. /// Returns when they all finish.
  1459. /// </summary>
  1460. /// <param name="actions">Actions to execute</param>
  1461. public void Join(params Action[] actions)
  1462. {
  1463. Join((IEnumerable<Action>)actions);
  1464. }
  1465. private class ChoiceIndex
  1466. {
  1467. public int _index = -1;
  1468. }
  1469. /// <summary>
  1470. /// Executes all actions in parallel
  1471. /// Returns when the first one completes
  1472. /// </summary>
  1473. /// <param name="actions">Actions to execute</param>
  1474. public int Choice(IEnumerable<Action> actions)
  1475. {
  1476. WIGStartInfo wigStartInfo = new WIGStartInfo { StartSuspended = true };
  1477. IWorkItemsGroup workItemsGroup = CreateWorkItemsGroup(int.MaxValue, wigStartInfo);
  1478. ManualResetEvent anActionCompleted = new ManualResetEvent(false);
  1479. ChoiceIndex choiceIndex = new ChoiceIndex();
  1480. int i = 0;
  1481. foreach (Action action in actions)
  1482. {
  1483. Action act = action;
  1484. int value = i;
  1485. workItemsGroup.QueueWorkItem(() => { act(); Interlocked.CompareExchange(ref choiceIndex._index, value, -1); anActionCompleted.Set(); });
  1486. ++i;
  1487. }
  1488. workItemsGroup.Start();
  1489. anActionCompleted.WaitOne();
  1490. anActionCompleted.Dispose();
  1491. return choiceIndex._index;
  1492. }
  1493. /// <summary>
  1494. /// Executes all actions in parallel
  1495. /// Returns when the first one completes
  1496. /// </summary>
  1497. /// <param name="actions">Actions to execute</param>
  1498. public int Choice(params Action[] actions)
  1499. {
  1500. return Choice((IEnumerable<Action>)actions);
  1501. }
  1502. /// <summary>
  1503. /// Executes actions in sequence asynchronously.
  1504. /// Returns immediately.
  1505. /// </summary>
  1506. /// <param name="pipeState">A state context that passes </param>
  1507. /// <param name="actions">Actions to execute in the order they should run</param>
  1508. public void Pipe<T>(T pipeState, IEnumerable<Action<T>> actions)
  1509. {
  1510. WIGStartInfo wigStartInfo = new WIGStartInfo { StartSuspended = true };
  1511. IWorkItemsGroup workItemsGroup = CreateWorkItemsGroup(1, wigStartInfo);
  1512. foreach (Action<T> action in actions)
  1513. {
  1514. Action<T> act = action;
  1515. workItemsGroup.QueueWorkItem(() => act(pipeState));
  1516. }
  1517. workItemsGroup.Start();
  1518. workItemsGroup.WaitForIdle();
  1519. }
  1520. /// <summary>
  1521. /// Executes actions in sequence asynchronously.
  1522. /// Returns immediately.
  1523. /// </summary>
  1524. /// <param name="pipeState"></param>
  1525. /// <param name="actions">Actions to execute in the order they should run</param>
  1526. public void Pipe<T>(T pipeState, params Action<T>[] actions)
  1527. {
  1528. Pipe(pipeState, (IEnumerable<Action<T>>)actions);
  1529. }
  1530. #endregion
  1531. }
  1532. #endregion
  1533. }