1
0

WorkItemsGroup.cs 9.2 KB

123456789101112131415161718192021222324252627282930313233343536373839404142434445464748495051525354555657585960616263646566676869707172737475767778798081828384858687888990919293949596979899100101102103104105106107108109110111112113114115116117118119120121122123124125126127128129130131132133134135136137138139140141142143144145146147148149150151152153154155156157158159160161162163164165166167168169170171172173174175176177178179180181182183184185186187188189190191192193194195196197198199200201202203204205206207208209210211212213214215216217218219220221222223224225226227228229230231232233234235236237238239240241242243244245246247248249250251252253254255256257258259260261262263264265266267268269270271272273274275276277278279280281282283284285286287288289290291292293294295296297298299300301302303304305306307308309310311312313314315316317318319320321322323324325326327328329330331332333334335336337338339340341342343344345346347348349350351352353354355356357358359360361
  1. using System;
  2. using System.Threading;
  3. using System.Runtime.CompilerServices;
  4. using System.Diagnostics;
  5. namespace Amib.Threading.Internal
  6. {
  7. #region WorkItemsGroup class
  8. /// <summary>
  9. /// Summary description for WorkItemsGroup.
  10. /// </summary>
  11. public class WorkItemsGroup : WorkItemsGroupBase
  12. {
  13. #region Private members
  14. private readonly object _lock = new object();
  15. /// <summary>
  16. /// A reference to the SmartThreadPool instance that created this
  17. /// WorkItemsGroup.
  18. /// </summary>
  19. private readonly SmartThreadPool _stp;
  20. /// <summary>
  21. /// The OnIdle event
  22. /// </summary>
  23. private event WorkItemsGroupIdleHandler _onIdle;
  24. /// <summary>
  25. /// A flag to indicate if the Work Items Group is now suspended.
  26. /// </summary>
  27. private bool _isSuspended;
  28. /// <summary>
  29. /// Defines how many work items of this WorkItemsGroup can run at once.
  30. /// </summary>
  31. private int _concurrency;
  32. /// <summary>
  33. /// Priority queue to hold work items before they are passed
  34. /// to the SmartThreadPool.
  35. /// </summary>
  36. private readonly PriorityQueue _workItemsQueue;
  37. /// <summary>
  38. /// Indicate how many work items are waiting in the SmartThreadPool
  39. /// queue.
  40. /// This value is used to apply the concurrency.
  41. /// </summary>
  42. private int _workItemsInStpQueue;
  43. /// <summary>
  44. /// Indicate how many work items are currently running in the SmartThreadPool.
  45. /// This value is used with the Cancel, to calculate if we can send new
  46. /// work items to the STP.
  47. /// </summary>
  48. private int _workItemsExecutingInStp = 0;
  49. /// <summary>
  50. /// WorkItemsGroup start information
  51. /// </summary>
  52. private readonly WIGStartInfo _workItemsGroupStartInfo;
  53. /// <summary>
  54. /// Signaled when all of the WorkItemsGroup's work item completed.
  55. /// </summary>
  56. //private readonly ManualResetEvent _isIdleWaitHandle = new ManualResetEvent(true);
  57. private readonly ManualResetEvent _isIdleWaitHandle = EventWaitHandleFactory.CreateManualResetEvent(true);
  58. /// <summary>
  59. /// A common object for all the work items that this work items group
  60. /// generate so we can mark them to cancel in O(1)
  61. /// </summary>
  62. private CanceledWorkItemsGroup _canceledWorkItemsGroup = new CanceledWorkItemsGroup();
  63. #endregion
  64. #region Construction
  65. public WorkItemsGroup(
  66. SmartThreadPool stp,
  67. int concurrency,
  68. WIGStartInfo wigStartInfo)
  69. {
  70. if (concurrency <= 0)
  71. {
  72. throw new ArgumentOutOfRangeException(
  73. "concurrency",
  74. #if !(_WINDOWS_CE) && !(_SILVERLIGHT) && !(WINDOWS_PHONE)
  75. concurrency,
  76. #endif
  77. "concurrency must be greater than zero");
  78. }
  79. _stp = stp;
  80. _concurrency = concurrency;
  81. _workItemsGroupStartInfo = new WIGStartInfo(wigStartInfo).AsReadOnly();
  82. _workItemsQueue = new PriorityQueue();
  83. Name = "WorkItemsGroup";
  84. // The _workItemsInStpQueue gets the number of currently executing work items,
  85. // because once a work item is executing, it cannot be cancelled.
  86. _workItemsInStpQueue = _workItemsExecutingInStp;
  87. _isSuspended = _workItemsGroupStartInfo.StartSuspended;
  88. }
  89. #endregion
  90. #region WorkItemsGroupBase Overrides
  91. public override int Concurrency
  92. {
  93. get { return _concurrency; }
  94. set
  95. {
  96. Debug.Assert(value > 0);
  97. int diff = value - _concurrency;
  98. _concurrency = value;
  99. if (diff > 0)
  100. {
  101. EnqueueToSTPNextNWorkItem(diff);
  102. }
  103. }
  104. }
  105. public override int WaitingCallbacks
  106. {
  107. get { return _workItemsQueue.Count; }
  108. }
  109. public override object[] GetStates()
  110. {
  111. lock (_lock)
  112. {
  113. object[] states = new object[_workItemsQueue.Count];
  114. int i = 0;
  115. foreach (WorkItem workItem in _workItemsQueue)
  116. {
  117. states[i] = workItem.GetWorkItemResult().State;
  118. ++i;
  119. }
  120. return states;
  121. }
  122. }
  123. /// <summary>
  124. /// WorkItemsGroup start information
  125. /// </summary>
  126. public override WIGStartInfo WIGStartInfo
  127. {
  128. get { return _workItemsGroupStartInfo; }
  129. }
  130. /// <summary>
  131. /// Start the Work Items Group if it was started suspended
  132. /// </summary>
  133. public override void Start()
  134. {
  135. // If the Work Items Group already started then quit
  136. if (!_isSuspended)
  137. {
  138. return;
  139. }
  140. _isSuspended = false;
  141. EnqueueToSTPNextNWorkItem(Math.Min(_workItemsQueue.Count, _concurrency));
  142. }
  143. public override void Cancel(bool abortExecution)
  144. {
  145. lock (_lock)
  146. {
  147. _canceledWorkItemsGroup.IsCanceled = true;
  148. _workItemsQueue.Clear();
  149. _workItemsInStpQueue = 0;
  150. _canceledWorkItemsGroup = new CanceledWorkItemsGroup();
  151. }
  152. if (abortExecution)
  153. {
  154. _stp.CancelAbortWorkItemsGroup(this);
  155. }
  156. }
  157. /// <summary>
  158. /// Wait for the thread pool to be idle
  159. /// </summary>
  160. public override bool WaitForIdle(int millisecondsTimeout)
  161. {
  162. SmartThreadPool.ValidateWorkItemsGroupWaitForIdle(this);
  163. return STPEventWaitHandle.WaitOne(_isIdleWaitHandle, millisecondsTimeout, false);
  164. }
  165. public override event WorkItemsGroupIdleHandler OnIdle
  166. {
  167. add { _onIdle += value; }
  168. remove { _onIdle -= value; }
  169. }
  170. #endregion
  171. #region Private methods
  172. private void RegisterToWorkItemCompletion(IWorkItemResult wir)
  173. {
  174. IInternalWorkItemResult iwir = (IInternalWorkItemResult)wir;
  175. iwir.OnWorkItemStarted += OnWorkItemStartedCallback;
  176. iwir.OnWorkItemCompleted += OnWorkItemCompletedCallback;
  177. }
  178. public void OnSTPIsStarting()
  179. {
  180. if (_isSuspended)
  181. {
  182. return;
  183. }
  184. EnqueueToSTPNextNWorkItem(_concurrency);
  185. }
  186. public void EnqueueToSTPNextNWorkItem(int count)
  187. {
  188. for (int i = 0; i < count; ++i)
  189. {
  190. EnqueueToSTPNextWorkItem(null, false);
  191. }
  192. }
  193. private object FireOnIdle(object state)
  194. {
  195. FireOnIdleImpl(_onIdle);
  196. return null;
  197. }
  198. [MethodImpl(MethodImplOptions.NoInlining)]
  199. private void FireOnIdleImpl(WorkItemsGroupIdleHandler onIdle)
  200. {
  201. if(null == onIdle)
  202. {
  203. return;
  204. }
  205. Delegate[] delegates = onIdle.GetInvocationList();
  206. foreach(WorkItemsGroupIdleHandler eh in delegates)
  207. {
  208. try
  209. {
  210. eh(this);
  211. }
  212. catch { } // Suppress exceptions
  213. }
  214. }
  215. private void OnWorkItemStartedCallback(WorkItem workItem)
  216. {
  217. lock(_lock)
  218. {
  219. ++_workItemsExecutingInStp;
  220. }
  221. }
  222. private void OnWorkItemCompletedCallback(WorkItem workItem)
  223. {
  224. EnqueueToSTPNextWorkItem(null, true);
  225. }
  226. internal override void Enqueue(WorkItem workItem)
  227. {
  228. EnqueueToSTPNextWorkItem(workItem);
  229. }
  230. private void EnqueueToSTPNextWorkItem(WorkItem workItem)
  231. {
  232. EnqueueToSTPNextWorkItem(workItem, false);
  233. }
  234. private void EnqueueToSTPNextWorkItem(WorkItem workItem, bool decrementWorkItemsInStpQueue)
  235. {
  236. lock(_lock)
  237. {
  238. // Got here from OnWorkItemCompletedCallback()
  239. if (decrementWorkItemsInStpQueue)
  240. {
  241. --_workItemsInStpQueue;
  242. if(_workItemsInStpQueue < 0)
  243. {
  244. _workItemsInStpQueue = 0;
  245. }
  246. --_workItemsExecutingInStp;
  247. if(_workItemsExecutingInStp < 0)
  248. {
  249. _workItemsExecutingInStp = 0;
  250. }
  251. }
  252. // If the work item is not null then enqueue it
  253. if (null != workItem)
  254. {
  255. workItem.CanceledWorkItemsGroup = _canceledWorkItemsGroup;
  256. RegisterToWorkItemCompletion(workItem.GetWorkItemResult());
  257. _workItemsQueue.Enqueue(workItem);
  258. //_stp.IncrementWorkItemsCount();
  259. if ((1 == _workItemsQueue.Count) &&
  260. (0 == _workItemsInStpQueue))
  261. {
  262. _stp.RegisterWorkItemsGroup(this);
  263. IsIdle = false;
  264. _isIdleWaitHandle.Reset();
  265. }
  266. }
  267. // If the work items queue of the group is empty than quit
  268. if (0 == _workItemsQueue.Count)
  269. {
  270. if (0 == _workItemsInStpQueue)
  271. {
  272. _stp.UnregisterWorkItemsGroup(this);
  273. IsIdle = true;
  274. _isIdleWaitHandle.Set();
  275. if (decrementWorkItemsInStpQueue && _onIdle != null && _onIdle.GetInvocationList().Length > 0)
  276. {
  277. _stp.QueueWorkItem(new WorkItemCallback(FireOnIdle));
  278. }
  279. }
  280. return;
  281. }
  282. if (!_isSuspended)
  283. {
  284. if (_workItemsInStpQueue < _concurrency)
  285. {
  286. WorkItem nextWorkItem = _workItemsQueue.Dequeue() as WorkItem;
  287. try
  288. {
  289. _stp.Enqueue(nextWorkItem);
  290. }
  291. catch (ObjectDisposedException e)
  292. {
  293. e.GetHashCode();
  294. // The STP has been shutdown
  295. }
  296. ++_workItemsInStpQueue;
  297. }
  298. }
  299. }
  300. }
  301. #endregion
  302. }
  303. #endregion
  304. }