aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/ThirdParty/SmartThreadPool/WorkItemsGroup.cs
diff options
context:
space:
mode:
authorTeravus Ovares2008-05-30 12:27:06 +0000
committerTeravus Ovares2008-05-30 12:27:06 +0000
commit1a47ff8094ee414a47aebd310826906d89428a09 (patch)
tree0e90b3a33f43ff8617a077bb57b86d6b28e63e71 /ThirdParty/SmartThreadPool/WorkItemsGroup.cs
parent* Fixed a dangling event hook that I added. (diff)
downloadopensim-SC_OLD-1a47ff8094ee414a47aebd310826906d89428a09.zip
opensim-SC_OLD-1a47ff8094ee414a47aebd310826906d89428a09.tar.gz
opensim-SC_OLD-1a47ff8094ee414a47aebd310826906d89428a09.tar.bz2
opensim-SC_OLD-1a47ff8094ee414a47aebd310826906d89428a09.tar.xz
* This is Melanie's XEngine script engine. I've not tested this real well, however, it's confirmed to compile and OpenSimulator to run successfully without this script engine active.
Diffstat (limited to '')
-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}