SmartThreadPool.cs 58 KB

1234567891011121314151617181920212223242526272829303132333435363738394041424344454647484950515253545556575859606162636465666768697071727374757677787980818283848586878889909192939495969798991001011021031041051061071081091101111121131141151161171181191201211221231241251261271281291301311321331341351361371381391401411421431441451461471481491501511521531541551561571581591601611621631641651661671681691701711721731741751761771781791801811821831841851861871881891901911921931941951961971981992002012022032042052062072082092102112122132142152162172182192202212222232242252262272282292302312322332342352362372382392402412422432442452462472482492502512522532542552562572582592602612622632642652662672682692702712722732742752762772782792802812822832842852862872882892902912922932942952962972982993003013023033043053063073083093103113123133143153163173183193203213223233243253263273283293303313323333343353363373383393403413423433443453463473483493503513523533543553563573583593603613623633643653663673683693703713723733743753763773783793803813823833843853863873883893903913923933943953963973983994004014024034044054064074084094104114124134144154164174184194204214224234244254264274284294304314324334344354364374384394404414424434444454464474484494504514524534544554564574584594604614624634644654664674684694704714724734744754764774784794804814824834844854864874884894904914924934944954964974984995005015025035045055065075085095105115125135145155165175185195205215225235245255265275285295305315325335345355365375385395405415425435445455465475485495505515525535545555565575585595605615625635645655665675685695705715725735745755765775785795805815825835845855865875885895905915925935945955965975985996006016026036046056066076086096106116126136146156166176186196206216226236246256266276286296306316326336346356366376386396406416426436446456466476486496506516526536546556566576586596606616626636646656666676686696706716726736746756766776786796806816826836846856866876886896906916926936946956966976986997007017027037047057067077087097107117127137147157167177187197207217227237247257267277287297307317327337347357367377387397407417427437447457467477487497507517527537547557567577587597607617627637647657667677687697707717727737747757767777787797807817827837847857867877887897907917927937947957967977987998008018028038048058068078088098108118128138148158168178188198208218228238248258268278288298308318328338348358368378388398408418428438448458468478488498508518528538548558568578588598608618628638648658668678688698708718728738748758768778788798808818828838848858868878888898908918928938948958968978988999009019029039049059069079089099109119129139149159169179189199209219229239249259269279289299309319329339349359369379389399409419429439449459469479489499509519529539549559569579589599609619629639649659669679689699709719729739749759769779789799809819829839849859869879889899909919929939949959969979989991000100110021003100410051006100710081009101010111012101310141015101610171018101910201021102210231024102510261027102810291030103110321033103410351036103710381039104010411042104310441045104610471048104910501051105210531054105510561057105810591060106110621063106410651066106710681069107010711072107310741075107610771078107910801081108210831084108510861087108810891090109110921093109410951096109710981099110011011102110311041105110611071108110911101111111211131114111511161117111811191120112111221123112411251126112711281129113011311132113311341135113611371138113911401141114211431144114511461147114811491150115111521153115411551156115711581159116011611162116311641165116611671168116911701171117211731174117511761177117811791180118111821183118411851186118711881189119011911192119311941195119611971198119912001201120212031204120512061207120812091210121112121213121412151216121712181219122012211222122312241225122612271228122912301231123212331234123512361237123812391240124112421243124412451246124712481249125012511252125312541255125612571258125912601261126212631264126512661267126812691270127112721273127412751276127712781279128012811282128312841285128612871288128912901291129212931294129512961297129812991300130113021303130413051306130713081309131013111312131313141315131613171318131913201321132213231324132513261327132813291330133113321333133413351336133713381339134013411342134313441345134613471348134913501351135213531354135513561357135813591360136113621363136413651366136713681369137013711372137313741375137613771378137913801381138213831384138513861387138813891390139113921393139413951396139713981399140014011402140314041405140614071408140914101411141214131414141514161417141814191420142114221423142414251426142714281429143014311432143314341435143614371438143914401441144214431444144514461447144814491450145114521453145414551456145714581459146014611462146314641465146614671468146914701471147214731474147514761477147814791480148114821483148414851486148714881489149014911492149314941495149614971498149915001501150215031504150515061507150815091510151115121513151415151516151715181519152015211522152315241525152615271528152915301531153215331534153515361537153815391540154115421543154415451546154715481549155015511552155315541555155615571558155915601561156215631564156515661567156815691570157115721573157415751576157715781579158015811582158315841585158615871588158915901591159215931594159515961597159815991600160116021603160416051606160716081609161016111612161316141615161616171618161916201621162216231624162516261627162816291630163116321633163416351636163716381639164016411642164316441645164616471648164916501651165216531654165516561657165816591660166116621663166416651666166716681669167016711672167316741675167616771678167916801681168216831684168516861687168816891690169116921693169416951696169716981699170017011702170317041705170617071708170917101711171217131714171517161717171817191720172117221723172417251726172717281729173017311732
  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.Start();
  604. workerThread.Name = string.Format("STP:{0}:{1}", Name, _threadCounter);
  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. _isDisposed = true;
  1314. }
  1315. }
  1316. private void ValidateNotDisposed()
  1317. {
  1318. if(_isDisposed)
  1319. {
  1320. throw new ObjectDisposedException(GetType().ToString(), "The SmartThreadPool has been shutdown");
  1321. }
  1322. }
  1323. #endregion
  1324. #region WorkItemsGroupBase Overrides
  1325. /// <summary>
  1326. /// Get/Set the maximum number of work items that execute cocurrency on the thread pool
  1327. /// </summary>
  1328. public override int Concurrency
  1329. {
  1330. get { return MaxThreads; }
  1331. set { MaxThreads = value; }
  1332. }
  1333. /// <summary>
  1334. /// Get the number of work items in the queue.
  1335. /// </summary>
  1336. public override int WaitingCallbacks
  1337. {
  1338. get
  1339. {
  1340. ValidateNotDisposed();
  1341. return _workItemsQueue.Count;
  1342. }
  1343. }
  1344. /// <summary>
  1345. /// Get an array with all the state objects of the currently running items.
  1346. /// The array represents a snap shot and impact performance.
  1347. /// </summary>
  1348. public override object[] GetStates()
  1349. {
  1350. object[] states = _workItemsQueue.GetStates();
  1351. return states;
  1352. }
  1353. /// <summary>
  1354. /// WorkItemsGroup start information (readonly)
  1355. /// </summary>
  1356. public override WIGStartInfo WIGStartInfo
  1357. {
  1358. get { return _stpStartInfo.AsReadOnly(); }
  1359. }
  1360. /// <summary>
  1361. /// Start the thread pool if it was started suspended.
  1362. /// If it is already running, this method is ignored.
  1363. /// </summary>
  1364. public override void Start()
  1365. {
  1366. if (!_isSuspended)
  1367. {
  1368. return;
  1369. }
  1370. _isSuspended = false;
  1371. ICollection workItemsGroups = _workItemsGroups.Values;
  1372. foreach (WorkItemsGroup workItemsGroup in workItemsGroups)
  1373. {
  1374. workItemsGroup.OnSTPIsStarting();
  1375. }
  1376. StartOptimalNumberOfThreads();
  1377. }
  1378. /// <summary>
  1379. /// Cancel all work items using thread abortion
  1380. /// </summary>
  1381. /// <param name="abortExecution">True to stop work items by raising ThreadAbortException</param>
  1382. public override void Cancel(bool abortExecution)
  1383. {
  1384. _canceledSmartThreadPool.IsCanceled = true;
  1385. _canceledSmartThreadPool = new CanceledWorkItemsGroup();
  1386. ICollection workItemsGroups = _workItemsGroups.Values;
  1387. foreach (WorkItemsGroup workItemsGroup in workItemsGroups)
  1388. {
  1389. workItemsGroup.Cancel(abortExecution);
  1390. }
  1391. if (abortExecution)
  1392. {
  1393. foreach (ThreadEntry threadEntry in _workerThreads.Values)
  1394. {
  1395. WorkItem workItem = threadEntry.CurrentWorkItem;
  1396. if (null != workItem &&
  1397. threadEntry.AssociatedSmartThreadPool == this &&
  1398. !workItem.IsCanceled)
  1399. {
  1400. threadEntry.CurrentWorkItem.GetWorkItemResult().Cancel(true);
  1401. }
  1402. }
  1403. }
  1404. }
  1405. /// <summary>
  1406. /// Wait for the thread pool to be idle
  1407. /// </summary>
  1408. public override bool WaitForIdle(int millisecondsTimeout)
  1409. {
  1410. ValidateWaitForIdle();
  1411. return STPEventWaitHandle.WaitOne(_isIdleWaitHandle, millisecondsTimeout, false);
  1412. }
  1413. /// <summary>
  1414. /// This event is fired when all work items are completed.
  1415. /// (When IsIdle changes to true)
  1416. /// This event only work on WorkItemsGroup. On SmartThreadPool
  1417. /// it throws the NotImplementedException.
  1418. /// </summary>
  1419. public override event WorkItemsGroupIdleHandler OnIdle
  1420. {
  1421. add
  1422. {
  1423. throw new NotImplementedException("This event is not implemented in the SmartThreadPool class. Please create a WorkItemsGroup in order to use this feature.");
  1424. //_onIdle += value;
  1425. }
  1426. remove
  1427. {
  1428. throw new NotImplementedException("This event is not implemented in the SmartThreadPool class. Please create a WorkItemsGroup in order to use this feature.");
  1429. //_onIdle -= value;
  1430. }
  1431. }
  1432. internal override void PreQueueWorkItem()
  1433. {
  1434. ValidateNotDisposed();
  1435. }
  1436. #endregion
  1437. #region Join, Choice, Pipe, etc.
  1438. /// <summary>
  1439. /// Executes all actions in parallel.
  1440. /// Returns when they all finish.
  1441. /// </summary>
  1442. /// <param name="actions">Actions to execute</param>
  1443. public void Join(IEnumerable<Action> actions)
  1444. {
  1445. WIGStartInfo wigStartInfo = new WIGStartInfo { StartSuspended = true };
  1446. IWorkItemsGroup workItemsGroup = CreateWorkItemsGroup(int.MaxValue, wigStartInfo);
  1447. foreach (Action action in actions)
  1448. {
  1449. workItemsGroup.QueueWorkItem(action);
  1450. }
  1451. workItemsGroup.Start();
  1452. workItemsGroup.WaitForIdle();
  1453. }
  1454. /// <summary>
  1455. /// Executes all actions in parallel.
  1456. /// Returns when they all finish.
  1457. /// </summary>
  1458. /// <param name="actions">Actions to execute</param>
  1459. public void Join(params Action[] actions)
  1460. {
  1461. Join((IEnumerable<Action>)actions);
  1462. }
  1463. private class ChoiceIndex
  1464. {
  1465. public int _index = -1;
  1466. }
  1467. /// <summary>
  1468. /// Executes all actions in parallel
  1469. /// Returns when the first one completes
  1470. /// </summary>
  1471. /// <param name="actions">Actions to execute</param>
  1472. public int Choice(IEnumerable<Action> actions)
  1473. {
  1474. WIGStartInfo wigStartInfo = new WIGStartInfo { StartSuspended = true };
  1475. IWorkItemsGroup workItemsGroup = CreateWorkItemsGroup(int.MaxValue, wigStartInfo);
  1476. ManualResetEvent anActionCompleted = new ManualResetEvent(false);
  1477. ChoiceIndex choiceIndex = new ChoiceIndex();
  1478. int i = 0;
  1479. foreach (Action action in actions)
  1480. {
  1481. Action act = action;
  1482. int value = i;
  1483. workItemsGroup.QueueWorkItem(() => { act(); Interlocked.CompareExchange(ref choiceIndex._index, value, -1); anActionCompleted.Set(); });
  1484. ++i;
  1485. }
  1486. workItemsGroup.Start();
  1487. anActionCompleted.WaitOne();
  1488. return choiceIndex._index;
  1489. }
  1490. /// <summary>
  1491. /// Executes all actions in parallel
  1492. /// Returns when the first one completes
  1493. /// </summary>
  1494. /// <param name="actions">Actions to execute</param>
  1495. public int Choice(params Action[] actions)
  1496. {
  1497. return Choice((IEnumerable<Action>)actions);
  1498. }
  1499. /// <summary>
  1500. /// Executes actions in sequence asynchronously.
  1501. /// Returns immediately.
  1502. /// </summary>
  1503. /// <param name="pipeState">A state context that passes </param>
  1504. /// <param name="actions">Actions to execute in the order they should run</param>
  1505. public void Pipe<T>(T pipeState, IEnumerable<Action<T>> actions)
  1506. {
  1507. WIGStartInfo wigStartInfo = new WIGStartInfo { StartSuspended = true };
  1508. IWorkItemsGroup workItemsGroup = CreateWorkItemsGroup(1, wigStartInfo);
  1509. foreach (Action<T> action in actions)
  1510. {
  1511. Action<T> act = action;
  1512. workItemsGroup.QueueWorkItem(() => act(pipeState));
  1513. }
  1514. workItemsGroup.Start();
  1515. workItemsGroup.WaitForIdle();
  1516. }
  1517. /// <summary>
  1518. /// Executes actions in sequence asynchronously.
  1519. /// Returns immediately.
  1520. /// </summary>
  1521. /// <param name="pipeState"></param>
  1522. /// <param name="actions">Actions to execute in the order they should run</param>
  1523. public void Pipe<T>(T pipeState, params Action<T>[] actions)
  1524. {
  1525. Pipe(pipeState, (IEnumerable<Action<T>>)actions);
  1526. }
  1527. #endregion
  1528. }
  1529. #endregion
  1530. }