aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/ThirdParty/SmartThreadPool/WorkItemsGroup.cs
diff options
context:
space:
mode:
Diffstat (limited to 'ThirdParty/SmartThreadPool/WorkItemsGroup.cs')
-rw-r--r--ThirdParty/SmartThreadPool/WorkItemsGroup.cs512
1 files changed, 512 insertions, 0 deletions
diff --git a/ThirdParty/SmartThreadPool/WorkItemsGroup.cs b/ThirdParty/SmartThreadPool/WorkItemsGroup.cs
new file mode 100644
index 0000000..01ac8dd
--- /dev/null
+++ b/ThirdParty/SmartThreadPool/WorkItemsGroup.cs
@@ -0,0 +1,512 @@
1// Ami Bar
2// amibar@gmail.com
3
4using System;
5using System.Threading;
6using System.Runtime.CompilerServices;
7using System.Diagnostics;
8
9namespace Amib.Threading.Internal
10{
11 #region WorkItemsGroup class
12
13 /// <summary>
14 /// Summary description for WorkItemsGroup.
15 /// </summary>
16 public class WorkItemsGroup : IWorkItemsGroup
17 {
18 #region Private members
19
20 private object _lock = new object();
21 /// <summary>
22 /// Contains the name of this instance of SmartThreadPool.
23 /// Can be changed by the user.
24 /// </summary>
25 private string _name = "WorkItemsGroup";
26
27 /// <summary>
28 /// A reference to the SmartThreadPool instance that created this
29 /// WorkItemsGroup.
30 /// </summary>
31 private SmartThreadPool _stp;
32
33 /// <summary>
34 /// The OnIdle event
35 /// </summary>
36 private event WorkItemsGroupIdleHandler _onIdle;
37
38 /// <summary>
39 /// Defines how many work items of this WorkItemsGroup can run at once.
40 /// </summary>
41 private int _concurrency;
42
43 /// <summary>
44 /// Priority queue to hold work items before they are passed
45 /// to the SmartThreadPool.
46 /// </summary>
47 private PriorityQueue _workItemsQueue;
48
49 /// <summary>
50 /// Indicate how many work items are waiting in the SmartThreadPool
51 /// queue.
52 /// This value is used to apply the concurrency.
53 /// </summary>
54 private int _workItemsInStpQueue;
55
56 /// <summary>
57 /// Indicate how many work items are currently running in the SmartThreadPool.
58 /// This value is used with the Cancel, to calculate if we can send new
59 /// work items to the STP.
60 /// </summary>
61 private int _workItemsExecutingInStp = 0;
62
63 /// <summary>
64 /// WorkItemsGroup start information
65 /// </summary>
66 private WIGStartInfo _workItemsGroupStartInfo;
67
68 /// <summary>
69 /// Signaled when all of the WorkItemsGroup's work item completed.
70 /// </summary>
71 private ManualResetEvent _isIdleWaitHandle = new ManualResetEvent(true);
72
73 /// <summary>
74 /// A common object for all the work items that this work items group
75 /// generate so we can mark them to cancel in O(1)
76 /// </summary>
77 private CanceledWorkItemsGroup _canceledWorkItemsGroup = new CanceledWorkItemsGroup();
78
79 #endregion
80
81 #region Construction
82
83 public WorkItemsGroup(
84 SmartThreadPool stp,
85 int concurrency,
86 WIGStartInfo wigStartInfo)
87 {
88 if (concurrency <= 0)
89 {
90 throw new ArgumentOutOfRangeException("concurrency", concurrency, "concurrency must be greater than zero");
91 }
92 _stp = stp;
93 _concurrency = concurrency;
94 _workItemsGroupStartInfo = new WIGStartInfo(wigStartInfo);
95 _workItemsQueue = new PriorityQueue();
96
97 // The _workItemsInStpQueue gets the number of currently executing work items,
98 // because once a work item is executing, it cannot be cancelled.
99 _workItemsInStpQueue = _workItemsExecutingInStp;
100 }
101
102 #endregion
103
104 #region IWorkItemsGroup implementation
105
106 /// <summary>
107 /// Get/Set the name of the SmartThreadPool instance
108 /// </summary>
109 public string Name
110 {
111 get
112 {
113 return _name;
114 }
115
116 set
117 {
118 _name = value;
119 }
120 }
121
122 /// <summary>
123 /// Queue a work item
124 /// </summary>
125 /// <param name="callback">A callback to execute</param>
126 /// <returns>Returns a work item result</returns>
127 public IWorkItemResult QueueWorkItem(WorkItemCallback callback)
128 {
129 WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _workItemsGroupStartInfo, callback);
130 EnqueueToSTPNextWorkItem(workItem);
131 return workItem.GetWorkItemResult();
132 }
133
134 /// <summary>
135 /// Queue a work item
136 /// </summary>
137 /// <param name="callback">A callback to execute</param>
138 /// <param name="workItemPriority">The priority of the work item</param>
139 /// <returns>Returns a work item result</returns>
140 public IWorkItemResult QueueWorkItem(WorkItemCallback callback, WorkItemPriority workItemPriority)
141 {
142 WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _workItemsGroupStartInfo, callback, workItemPriority);
143 EnqueueToSTPNextWorkItem(workItem);
144 return workItem.GetWorkItemResult();
145 }
146
147 /// <summary>
148 /// Queue a work item
149 /// </summary>
150 /// <param name="workItemInfo">Work item info</param>
151 /// <param name="callback">A callback to execute</param>
152 /// <returns>Returns a work item result</returns>
153 public IWorkItemResult QueueWorkItem(WorkItemInfo workItemInfo, WorkItemCallback callback)
154 {
155 WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _workItemsGroupStartInfo, workItemInfo, callback);
156 EnqueueToSTPNextWorkItem(workItem);
157 return workItem.GetWorkItemResult();
158 }
159
160 /// <summary>
161 /// Queue a work item
162 /// </summary>
163 /// <param name="callback">A callback to execute</param>
164 /// <param name="state">
165 /// The context object of the work item. Used for passing arguments to the work item.
166 /// </param>
167 /// <returns>Returns a work item result</returns>
168 public IWorkItemResult QueueWorkItem(WorkItemCallback callback, object state)
169 {
170 WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _workItemsGroupStartInfo, callback, state);
171 EnqueueToSTPNextWorkItem(workItem);
172 return workItem.GetWorkItemResult();
173 }
174
175 /// <summary>
176 /// Queue a work item
177 /// </summary>
178 /// <param name="callback">A callback to execute</param>
179 /// <param name="state">
180 /// The context object of the work item. Used for passing arguments to the work item.
181 /// </param>
182 /// <param name="workItemPriority">The work item priority</param>
183 /// <returns>Returns a work item result</returns>
184 public IWorkItemResult QueueWorkItem(WorkItemCallback callback, object state, WorkItemPriority workItemPriority)
185 {
186 WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _workItemsGroupStartInfo, callback, state, workItemPriority);
187 EnqueueToSTPNextWorkItem(workItem);
188 return workItem.GetWorkItemResult();
189 }
190
191 /// <summary>
192 /// Queue a work item
193 /// </summary>
194 /// <param name="workItemInfo">Work item information</param>
195 /// <param name="callback">A callback to execute</param>
196 /// <param name="state">
197 /// The context object of the work item. Used for passing arguments to the work item.
198 /// </param>
199 /// <returns>Returns a work item result</returns>
200 public IWorkItemResult QueueWorkItem(WorkItemInfo workItemInfo, WorkItemCallback callback, object state)
201 {
202 WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _workItemsGroupStartInfo, workItemInfo, callback, state);
203 EnqueueToSTPNextWorkItem(workItem);
204 return workItem.GetWorkItemResult();
205 }
206
207 /// <summary>
208 /// Queue a work item
209 /// </summary>
210 /// <param name="callback">A callback to execute</param>
211 /// <param name="state">
212 /// The context object of the work item. Used for passing arguments to the work item.
213 /// </param>
214 /// <param name="postExecuteWorkItemCallback">
215 /// A delegate to call after the callback completion
216 /// </param>
217 /// <returns>Returns a work item result</returns>
218 public IWorkItemResult QueueWorkItem(
219 WorkItemCallback callback,
220 object state,
221 PostExecuteWorkItemCallback postExecuteWorkItemCallback)
222 {
223 WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _workItemsGroupStartInfo, callback, state, postExecuteWorkItemCallback);
224 EnqueueToSTPNextWorkItem(workItem);
225 return workItem.GetWorkItemResult();
226 }
227
228 /// <summary>
229 /// Queue a work item
230 /// </summary>
231 /// <param name="callback">A callback to execute</param>
232 /// <param name="state">
233 /// The context object of the work item. Used for passing arguments to the work item.
234 /// </param>
235 /// <param name="postExecuteWorkItemCallback">
236 /// A delegate to call after the callback completion
237 /// </param>
238 /// <param name="workItemPriority">The work item priority</param>
239 /// <returns>Returns a work item result</returns>
240 public IWorkItemResult QueueWorkItem(
241 WorkItemCallback callback,
242 object state,
243 PostExecuteWorkItemCallback postExecuteWorkItemCallback,
244 WorkItemPriority workItemPriority)
245 {
246 WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _workItemsGroupStartInfo, callback, state, postExecuteWorkItemCallback, workItemPriority);
247 EnqueueToSTPNextWorkItem(workItem);
248 return workItem.GetWorkItemResult();
249 }
250
251 /// <summary>
252 /// Queue a work item
253 /// </summary>
254 /// <param name="callback">A callback to execute</param>
255 /// <param name="state">
256 /// The context object of the work item. Used for passing arguments to the work item.
257 /// </param>
258 /// <param name="postExecuteWorkItemCallback">
259 /// A delegate to call after the callback completion
260 /// </param>
261 /// <param name="callToPostExecute">Indicates on which cases to call to the post execute callback</param>
262 /// <returns>Returns a work item result</returns>
263 public IWorkItemResult QueueWorkItem(
264 WorkItemCallback callback,
265 object state,
266 PostExecuteWorkItemCallback postExecuteWorkItemCallback,
267 CallToPostExecute callToPostExecute)
268 {
269 WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _workItemsGroupStartInfo, callback, state, postExecuteWorkItemCallback, callToPostExecute);
270 EnqueueToSTPNextWorkItem(workItem);
271 return workItem.GetWorkItemResult();
272 }
273
274 /// <summary>
275 /// Queue a work item
276 /// </summary>
277 /// <param name="callback">A callback to execute</param>
278 /// <param name="state">
279 /// The context object of the work item. Used for passing arguments to the work item.
280 /// </param>
281 /// <param name="postExecuteWorkItemCallback">
282 /// A delegate to call after the callback completion
283 /// </param>
284 /// <param name="callToPostExecute">Indicates on which cases to call to the post execute callback</param>
285 /// <param name="workItemPriority">The work item priority</param>
286 /// <returns>Returns a work item result</returns>
287 public IWorkItemResult QueueWorkItem(
288 WorkItemCallback callback,
289 object state,
290 PostExecuteWorkItemCallback postExecuteWorkItemCallback,
291 CallToPostExecute callToPostExecute,
292 WorkItemPriority workItemPriority)
293 {
294 WorkItem workItem = WorkItemFactory.CreateWorkItem(this, _workItemsGroupStartInfo, callback, state, postExecuteWorkItemCallback, callToPostExecute, workItemPriority);
295 EnqueueToSTPNextWorkItem(workItem);
296 return workItem.GetWorkItemResult();
297 }
298
299 /// <summary>
300 /// Wait for the thread pool to be idle
301 /// </summary>
302 public void WaitForIdle()
303 {
304 WaitForIdle(Timeout.Infinite);
305 }
306
307 /// <summary>
308 /// Wait for the thread pool to be idle
309 /// </summary>
310 public bool WaitForIdle(TimeSpan timeout)
311 {
312 return WaitForIdle((int)timeout.TotalMilliseconds);
313 }
314
315 /// <summary>
316 /// Wait for the thread pool to be idle
317 /// </summary>
318 public bool WaitForIdle(int millisecondsTimeout)
319 {
320 _stp.ValidateWorkItemsGroupWaitForIdle(this);
321 return _isIdleWaitHandle.WaitOne(millisecondsTimeout, false);
322 }
323
324 public int WaitingCallbacks
325 {
326 get
327 {
328 return _workItemsQueue.Count;
329 }
330 }
331
332 public event WorkItemsGroupIdleHandler OnIdle
333 {
334 add
335 {
336 _onIdle += value;
337 }
338 remove
339 {
340 _onIdle -= value;
341 }
342 }
343
344 public void Cancel()
345 {
346 lock(_lock)
347 {
348 _canceledWorkItemsGroup.IsCanceled = true;
349 _workItemsQueue.Clear();
350 _workItemsInStpQueue = 0;
351 _canceledWorkItemsGroup = new CanceledWorkItemsGroup();
352 }
353 }
354
355 public void Start()
356 {
357 lock (this)
358 {
359 if (!_workItemsGroupStartInfo.StartSuspended)
360 {
361 return;
362 }
363 _workItemsGroupStartInfo.StartSuspended = false;
364 }
365
366 for(int i = 0; i < _concurrency; ++i)
367 {
368 EnqueueToSTPNextWorkItem(null, false);
369 }
370 }
371
372 #endregion
373
374 #region Private methods
375
376 private void RegisterToWorkItemCompletion(IWorkItemResult wir)
377 {
378 IInternalWorkItemResult iwir = wir as IInternalWorkItemResult;
379 iwir.OnWorkItemStarted += new WorkItemStateCallback(OnWorkItemStartedCallback);
380 iwir.OnWorkItemCompleted += new WorkItemStateCallback(OnWorkItemCompletedCallback);
381 }
382
383 public void OnSTPIsStarting()
384 {
385 lock (this)
386 {
387 if (_workItemsGroupStartInfo.StartSuspended)
388 {
389 return;
390 }
391 }
392
393 for(int i = 0; i < _concurrency; ++i)
394 {
395 EnqueueToSTPNextWorkItem(null, false);
396 }
397 }
398
399 private object FireOnIdle(object state)
400 {
401 FireOnIdleImpl(_onIdle);
402 return null;
403 }
404
405 [MethodImpl(MethodImplOptions.NoInlining)]
406 private void FireOnIdleImpl(WorkItemsGroupIdleHandler onIdle)
407 {
408 if(null == onIdle)
409 {
410 return;
411 }
412
413 Delegate[] delegates = onIdle.GetInvocationList();
414 foreach(WorkItemsGroupIdleHandler eh in delegates)
415 {
416 try
417 {
418 eh(this);
419 }
420 // Ignore exceptions
421 catch{}
422 }
423 }
424
425 private void OnWorkItemStartedCallback(WorkItem workItem)
426 {
427 lock(_lock)
428 {
429 ++_workItemsExecutingInStp;
430 }
431 }
432
433 private void OnWorkItemCompletedCallback(WorkItem workItem)
434 {
435 EnqueueToSTPNextWorkItem(null, true);
436 }
437
438 private void EnqueueToSTPNextWorkItem(WorkItem workItem)
439 {
440 EnqueueToSTPNextWorkItem(workItem, false);
441 }
442
443 private void EnqueueToSTPNextWorkItem(WorkItem workItem, bool decrementWorkItemsInStpQueue)
444 {
445 lock(_lock)
446 {
447 // Got here from OnWorkItemCompletedCallback()
448 if (decrementWorkItemsInStpQueue)
449 {
450 --_workItemsInStpQueue;
451
452 if(_workItemsInStpQueue < 0)
453 {
454 _workItemsInStpQueue = 0;
455 }
456
457 --_workItemsExecutingInStp;
458
459 if(_workItemsExecutingInStp < 0)
460 {
461 _workItemsExecutingInStp = 0;
462 }
463 }
464
465 // If the work item is not null then enqueue it
466 if (null != workItem)
467 {
468 workItem.CanceledWorkItemsGroup = _canceledWorkItemsGroup;
469
470 RegisterToWorkItemCompletion(workItem.GetWorkItemResult());
471 _workItemsQueue.Enqueue(workItem);
472 //_stp.IncrementWorkItemsCount();
473
474 if ((1 == _workItemsQueue.Count) &&
475 (0 == _workItemsInStpQueue))
476 {
477 _stp.RegisterWorkItemsGroup(this);
478 Trace.WriteLine("WorkItemsGroup " + Name + " is NOT idle");
479 _isIdleWaitHandle.Reset();
480 }
481 }
482
483 // If the work items queue of the group is empty than quit
484 if (0 == _workItemsQueue.Count)
485 {
486 if (0 == _workItemsInStpQueue)
487 {
488 _stp.UnregisterWorkItemsGroup(this);
489 Trace.WriteLine("WorkItemsGroup " + Name + " is idle");
490 _isIdleWaitHandle.Set();
491 _stp.QueueWorkItem(new WorkItemCallback(this.FireOnIdle));
492 }
493 return;
494 }
495
496 if (!_workItemsGroupStartInfo.StartSuspended)
497 {
498 if (_workItemsInStpQueue < _concurrency)
499 {
500 WorkItem nextWorkItem = _workItemsQueue.Dequeue() as WorkItem;
501 _stp.Enqueue(nextWorkItem, true);
502 ++_workItemsInStpQueue;
503 }
504 }
505 }
506 }
507
508 #endregion
509 }
510
511 #endregion
512}