aboutsummaryrefslogtreecommitdiffstatshomepage
path: root/ThirdParty/SmartThreadPool/WorkItemsQueue.cs
diff options
context:
space:
mode:
Diffstat (limited to 'ThirdParty/SmartThreadPool/WorkItemsQueue.cs')
-rw-r--r--ThirdParty/SmartThreadPool/WorkItemsQueue.cs1245
1 files changed, 645 insertions, 600 deletions
diff --git a/ThirdParty/SmartThreadPool/WorkItemsQueue.cs b/ThirdParty/SmartThreadPool/WorkItemsQueue.cs
index af5af07..156a131 100644
--- a/ThirdParty/SmartThreadPool/WorkItemsQueue.cs
+++ b/ThirdParty/SmartThreadPool/WorkItemsQueue.cs
@@ -1,600 +1,645 @@
1// Ami Bar 1using System;
2// amibar@gmail.com 2using System.Collections.Generic;
3 3using System.Threading;
4using System; 4
5using System.Threading; 5namespace Amib.Threading.Internal
6 6{
7namespace Amib.Threading.Internal 7 #region WorkItemsQueue class
8{ 8
9 #region WorkItemsQueue class 9 /// <summary>
10 10 /// WorkItemsQueue class.
11 /// <summary> 11 /// </summary>
12 /// WorkItemsQueue class. 12 public class WorkItemsQueue : IDisposable
13 /// </summary> 13 {
14 public class WorkItemsQueue : IDisposable 14 #region Member variables
15 { 15
16 #region Member variables 16 /// <summary>
17 17 /// Waiters queue (implemented as stack).
18 /// <summary> 18 /// </summary>
19 /// Waiters queue (implemented as stack). 19 private readonly WaiterEntry _headWaiterEntry = new WaiterEntry();
20 /// </summary> 20
21 private WaiterEntry _headWaiterEntry = new WaiterEntry(); 21 /// <summary>
22 22 /// Waiters count
23 /// <summary> 23 /// </summary>
24 /// Waiters count 24 private int _waitersCount = 0;
25 /// </summary> 25
26 private int _waitersCount = 0; 26 /// <summary>
27 27 /// Work items queue
28 /// <summary> 28 /// </summary>
29 /// Work items queue 29 private readonly PriorityQueue _workItems = new PriorityQueue();
30 /// </summary> 30
31 private PriorityQueue _workItems = new PriorityQueue(); 31 /// <summary>
32 32 /// Indicate that work items are allowed to be queued
33 /// <summary> 33 /// </summary>
34 /// Indicate that work items are allowed to be queued 34 private bool _isWorkItemsQueueActive = true;
35 /// </summary> 35
36 private bool _isWorkItemsQueueActive = true; 36
37 37#if (WINDOWS_PHONE)
38 /// <summary> 38 private static readonly Dictionary<int, WaiterEntry> _waiterEntries = new Dictionary<int, WaiterEntry>();
39 /// Each thread in the thread pool keeps its own waiter entry. 39#elif (_WINDOWS_CE)
40 /// </summary> 40 private static LocalDataStoreSlot _waiterEntrySlot = Thread.AllocateDataSlot();
41 [ThreadStatic] 41#else
42 private static WaiterEntry _waiterEntry; 42
43 43 [ThreadStatic]
44 /// <summary> 44 private static WaiterEntry _waiterEntry;
45 /// A flag that indicates if the WorkItemsQueue has been disposed. 45#endif
46 /// </summary> 46
47 private bool _isDisposed = false; 47
48 48 /// <summary>
49 #endregion 49 /// Each thread in the thread pool keeps its own waiter entry.
50 50 /// </summary>
51 #region Public properties 51 private static WaiterEntry CurrentWaiterEntry
52 52 {
53 /// <summary> 53#if (WINDOWS_PHONE)
54 /// Returns the current number of work items in the queue 54 get
55 /// </summary> 55 {
56 public int Count 56 lock (_waiterEntries)
57 { 57 {
58 get 58 WaiterEntry waiterEntry;
59 { 59 if (_waiterEntries.TryGetValue(Thread.CurrentThread.ManagedThreadId, out waiterEntry))
60 lock(this) 60 {
61 { 61 return waiterEntry;
62 ValidateNotDisposed(); 62 }
63 return _workItems.Count; 63 }
64 } 64 return null;
65 } 65 }
66 } 66 set
67 67 {
68 /// <summary> 68 lock (_waiterEntries)
69 /// Returns the current number of waiters 69 {
70 /// </summary> 70 _waiterEntries[Thread.CurrentThread.ManagedThreadId] = value;
71 public int WaitersCount 71 }
72 { 72 }
73 get 73#elif (_WINDOWS_CE)
74 { 74 get
75 lock(this) 75 {
76 { 76 return Thread.GetData(_waiterEntrySlot) as WaiterEntry;
77 ValidateNotDisposed(); 77 }
78 return _waitersCount; 78 set
79 } 79 {
80 } 80 Thread.SetData(_waiterEntrySlot, value);
81 } 81 }
82 82#else
83 83 get
84 #endregion 84 {
85 85 return _waiterEntry;
86 #region Public methods 86 }
87 87 set
88 /// <summary> 88 {
89 /// Enqueue a work item to the queue. 89 _waiterEntry = value;
90 /// </summary> 90 }
91 public bool EnqueueWorkItem(WorkItem workItem) 91#endif
92 { 92 }
93 // A work item cannot be null, since null is used in the 93
94 // WaitForWorkItem() method to indicate timeout or cancel 94 /// <summary>
95 if (null == workItem) 95 /// A flag that indicates if the WorkItemsQueue has been disposed.
96 { 96 /// </summary>
97 throw new ArgumentNullException("workItem" , "workItem cannot be null"); 97 private bool _isDisposed = false;
98 } 98
99 99 #endregion
100 bool enqueue = true; 100
101 101 #region Public properties
102 // First check if there is a waiter waiting for work item. During 102
103 // the check, timed out waiters are ignored. If there is no 103 /// <summary>
104 // waiter then the work item is queued. 104 /// Returns the current number of work items in the queue
105 lock(this) 105 /// </summary>
106 { 106 public int Count
107 ValidateNotDisposed(); 107 {
108 108 get
109 if (!_isWorkItemsQueueActive) 109 {
110 { 110 return _workItems.Count;
111 return false; 111 }
112 } 112 }
113 113
114 while(_waitersCount > 0) 114 /// <summary>
115 { 115 /// Returns the current number of waiters
116 // Dequeue a waiter. 116 /// </summary>
117 WaiterEntry waiterEntry = PopWaiter(); 117 public int WaitersCount
118 118 {
119 // Signal the waiter. On success break the loop 119 get
120 if (waiterEntry.Signal(workItem)) 120 {
121 { 121 return _waitersCount;
122 enqueue = false; 122 }
123 break; 123 }
124 } 124
125 } 125
126 126 #endregion
127 if (enqueue) 127
128 { 128 #region Public methods
129 // Enqueue the work item 129
130 _workItems.Enqueue(workItem); 130 /// <summary>
131 } 131 /// Enqueue a work item to the queue.
132 } 132 /// </summary>
133 return true; 133 public bool EnqueueWorkItem(WorkItem workItem)
134 } 134 {
135 135 // A work item cannot be null, since null is used in the
136 136 // WaitForWorkItem() method to indicate timeout or cancel
137 /// <summary> 137 if (null == workItem)
138 /// Waits for a work item or exits on timeout or cancel 138 {
139 /// </summary> 139 throw new ArgumentNullException("workItem" , "workItem cannot be null");
140 /// <param name="millisecondsTimeout">Timeout in milliseconds</param> 140 }
141 /// <param name="cancelEvent">Cancel wait handle</param> 141
142 /// <returns>Returns true if the resource was granted</returns> 142 bool enqueue = true;
143 public WorkItem DequeueWorkItem( 143
144 int millisecondsTimeout, 144 // First check if there is a waiter waiting for work item. During
145 WaitHandle cancelEvent) 145 // the check, timed out waiters are ignored. If there is no
146 { 146 // waiter then the work item is queued.
147 /// This method cause the caller to wait for a work item. 147 lock(this)
148 /// If there is at least one waiting work item then the 148 {
149 /// method returns immidiately with true. 149 ValidateNotDisposed();
150 /// 150
151 /// If there are no waiting work items then the caller 151 if (!_isWorkItemsQueueActive)
152 /// is queued between other waiters for a work item to arrive. 152 {
153 /// 153 return false;
154 /// If a work item didn't come within millisecondsTimeout or 154 }
155 /// the user canceled the wait by signaling the cancelEvent 155
156 /// then the method returns false to indicate that the caller 156 while(_waitersCount > 0)
157 /// didn't get a work item. 157 {
158 158 // Dequeue a waiter.
159 WaiterEntry waiterEntry = null; 159 WaiterEntry waiterEntry = PopWaiter();
160 WorkItem workItem = null; 160
161 161 // Signal the waiter. On success break the loop
162 lock(this) 162 if (waiterEntry.Signal(workItem))
163 { 163 {
164 ValidateNotDisposed(); 164 enqueue = false;
165 165 break;
166 // If there are waiting work items then take one and return. 166 }
167 if (_workItems.Count > 0) 167 }
168 { 168
169 workItem = _workItems.Dequeue() as WorkItem; 169 if (enqueue)
170 return workItem; 170 {
171 } 171 // Enqueue the work item
172 // No waiting work items ... 172 _workItems.Enqueue(workItem);
173 else 173 }
174 { 174 }
175 // Get the wait entry for the waiters queue 175 return true;
176 waiterEntry = GetThreadWaiterEntry(); 176 }
177 177
178 // Put the waiter with the other waiters 178
179 PushWaiter(waiterEntry); 179 /// <summary>
180 } 180 /// Waits for a work item or exits on timeout or cancel
181 } 181 /// </summary>
182 182 /// <param name="millisecondsTimeout">Timeout in milliseconds</param>
183 // Prepare array of wait handle for the WaitHandle.WaitAny() 183 /// <param name="cancelEvent">Cancel wait handle</param>
184 WaitHandle [] waitHandles = new WaitHandle [] { 184 /// <returns>Returns true if the resource was granted</returns>
185 waiterEntry.WaitHandle, 185 public WorkItem DequeueWorkItem(
186 cancelEvent }; 186 int millisecondsTimeout,
187 187 WaitHandle cancelEvent)
188 // Wait for an available resource, cancel event, or timeout. 188 {
189 189 // This method cause the caller to wait for a work item.
190 // During the wait we are supposes to exit the synchronization 190 // If there is at least one waiting work item then the
191 // domain. (Placing true as the third argument of the WaitAny()) 191 // method returns immidiately with it.
192 // It just doesn't work, I don't know why, so I have lock(this) 192 //
193 // statments insted of one. 193 // If there are no waiting work items then the caller
194 194 // is queued between other waiters for a work item to arrive.
195 int index = WaitHandle.WaitAny( 195 //
196 waitHandles, 196 // If a work item didn't come within millisecondsTimeout or
197 millisecondsTimeout, 197 // the user canceled the wait by signaling the cancelEvent
198 true); 198 // then the method returns null to indicate that the caller
199 199 // didn't get a work item.
200 lock(this) 200
201 { 201 WaiterEntry waiterEntry;
202 // success is true if it got a work item. 202 WorkItem workItem = null;
203 bool success = (0 == index); 203 lock (this)
204 204 {
205 // The timeout variable is used only for readability. 205 ValidateNotDisposed();
206 // (We treat cancel as timeout) 206
207 bool timeout = !success; 207 // If there are waiting work items then take one and return.
208 208 if (_workItems.Count > 0)
209 // On timeout update the waiterEntry that it is timed out 209 {
210 if (timeout) 210 workItem = _workItems.Dequeue() as WorkItem;
211 { 211 return workItem;
212 // The Timeout() fails if the waiter has already been signaled 212 }
213 timeout = waiterEntry.Timeout(); 213
214 214 // No waiting work items ...
215 // On timeout remove the waiter from the queue. 215
216 // Note that the complexity is O(1). 216 // Get the waiter entry for the waiters queue
217 if(timeout) 217 waiterEntry = GetThreadWaiterEntry();
218 { 218
219 RemoveWaiter(waiterEntry, false); 219 // Put the waiter with the other waiters
220 } 220 PushWaiter(waiterEntry);
221 221 }
222 // Again readability 222
223 success = !timeout; 223 // Prepare array of wait handle for the WaitHandle.WaitAny()
224 } 224 WaitHandle [] waitHandles = new WaitHandle[] {
225 225 waiterEntry.WaitHandle,
226 // On success return the work item 226 cancelEvent };
227 if (success) 227
228 { 228 // Wait for an available resource, cancel event, or timeout.
229 workItem = waiterEntry.WorkItem; 229
230 230 // During the wait we are supposes to exit the synchronization
231 if (null == workItem) 231 // domain. (Placing true as the third argument of the WaitAny())
232 { 232 // It just doesn't work, I don't know why, so I have two lock(this)
233 workItem = _workItems.Dequeue() as WorkItem; 233 // statments instead of one.
234 } 234
235 } 235 int index = STPEventWaitHandle.WaitAny(
236 } 236 waitHandles,
237 // On failure return null. 237 millisecondsTimeout,
238 return workItem; 238 true);
239 } 239
240 240 lock(this)
241 /// <summary> 241 {
242 /// Cleanup the work items queue, hence no more work 242 // success is true if it got a work item.
243 /// items are allowed to be queue 243 bool success = (0 == index);
244 /// </summary> 244
245 protected virtual void Cleanup() 245 // The timeout variable is used only for readability.
246 { 246 // (We treat cancel as timeout)
247 lock(this) 247 bool timeout = !success;
248 { 248
249 // Deactivate only once 249 // On timeout update the waiterEntry that it is timed out
250 if (!_isWorkItemsQueueActive) 250 if (timeout)
251 { 251 {
252 return; 252 // The Timeout() fails if the waiter has already been signaled
253 } 253 timeout = waiterEntry.Timeout();
254 254
255 // Don't queue more work items 255 // On timeout remove the waiter from the queue.
256 _isWorkItemsQueueActive = false; 256 // Note that the complexity is O(1).
257 257 if(timeout)
258 foreach(WorkItem workItem in _workItems) 258 {
259 { 259 RemoveWaiter(waiterEntry, false);
260 workItem.DisposeOfState(); 260 }
261 } 261
262 262 // Again readability
263 // Clear the work items that are already queued 263 success = !timeout;
264 _workItems.Clear(); 264 }
265 265
266 // Note: 266 // On success return the work item
267 // I don't iterate over the queue and dispose of work items's states, 267 if (success)
268 // since if a work item has a state object that is still in use in the 268 {
269 // application then I must not dispose it. 269 workItem = waiterEntry.WorkItem;
270 270
271 // Tell the waiters that they were timed out. 271 if (null == workItem)
272 // It won't signal them to exit, but to ignore their 272 {
273 // next work item. 273 workItem = _workItems.Dequeue() as WorkItem;
274 while(_waitersCount > 0) 274 }
275 { 275 }
276 WaiterEntry waiterEntry = PopWaiter(); 276 }
277 waiterEntry.Timeout(); 277 // On failure return null.
278 } 278 return workItem;
279 } 279 }
280 } 280
281 281 /// <summary>
282 #endregion 282 /// Cleanup the work items queue, hence no more work
283 283 /// items are allowed to be queue
284 #region Private methods 284 /// </summary>
285 285 private void Cleanup()
286 /// <summary> 286 {
287 /// Returns the WaiterEntry of the current thread 287 lock(this)
288 /// </summary> 288 {
289 /// <returns></returns> 289 // Deactivate only once
290 /// In order to avoid creation and destuction of WaiterEntry 290 if (!_isWorkItemsQueueActive)
291 /// objects each thread has its own WaiterEntry object. 291 {
292 private WaiterEntry GetThreadWaiterEntry() 292 return;
293 { 293 }
294 if (null == _waiterEntry) 294
295 { 295 // Don't queue more work items
296 _waiterEntry = new WaiterEntry(); 296 _isWorkItemsQueueActive = false;
297 } 297
298 _waiterEntry.Reset(); 298 foreach(WorkItem workItem in _workItems)
299 return _waiterEntry; 299 {
300 } 300 workItem.DisposeOfState();
301 301 }
302 #region Waiters stack methods 302
303 303 // Clear the work items that are already queued
304 /// <summary> 304 _workItems.Clear();
305 /// Push a new waiter into the waiter's stack 305
306 /// </summary> 306 // Note:
307 /// <param name="newWaiterEntry">A waiter to put in the stack</param> 307 // I don't iterate over the queue and dispose of work items's states,
308 public void PushWaiter(WaiterEntry newWaiterEntry) 308 // since if a work item has a state object that is still in use in the
309 { 309 // application then I must not dispose it.
310 // Remove the waiter if it is already in the stack and 310
311 // update waiter's count as needed 311 // Tell the waiters that they were timed out.
312 RemoveWaiter(newWaiterEntry, false); 312 // It won't signal them to exit, but to ignore their
313 313 // next work item.
314 // If the stack is empty then newWaiterEntry is the new head of the stack 314 while(_waitersCount > 0)
315 if (null == _headWaiterEntry._nextWaiterEntry) 315 {
316 { 316 WaiterEntry waiterEntry = PopWaiter();
317 _headWaiterEntry._nextWaiterEntry = newWaiterEntry; 317 waiterEntry.Timeout();
318 newWaiterEntry._prevWaiterEntry = _headWaiterEntry; 318 }
319 319 }
320 } 320 }
321 // If the stack is not empty then put newWaiterEntry as the new head 321
322 // of the stack. 322 public object[] GetStates()
323 else 323 {
324 { 324 lock (this)
325 // Save the old first waiter entry 325 {
326 WaiterEntry oldFirstWaiterEntry = _headWaiterEntry._nextWaiterEntry; 326 object[] states = new object[_workItems.Count];
327 327 int i = 0;
328 // Update the links 328 foreach (WorkItem workItem in _workItems)
329 _headWaiterEntry._nextWaiterEntry = newWaiterEntry; 329 {
330 newWaiterEntry._nextWaiterEntry = oldFirstWaiterEntry; 330 states[i] = workItem.GetWorkItemResult().State;
331 newWaiterEntry._prevWaiterEntry = _headWaiterEntry; 331 ++i;
332 oldFirstWaiterEntry._prevWaiterEntry = newWaiterEntry; 332 }
333 } 333 return states;
334 334 }
335 // Increment the number of waiters 335 }
336 ++_waitersCount; 336
337 } 337 #endregion
338 338
339 /// <summary> 339 #region Private methods
340 /// Pop a waiter from the waiter's stack 340
341 /// </summary> 341 /// <summary>
342 /// <returns>Returns the first waiter in the stack</returns> 342 /// Returns the WaiterEntry of the current thread
343 private WaiterEntry PopWaiter() 343 /// </summary>
344 { 344 /// <returns></returns>
345 // Store the current stack head 345 /// In order to avoid creation and destuction of WaiterEntry
346 WaiterEntry oldFirstWaiterEntry = _headWaiterEntry._nextWaiterEntry; 346 /// objects each thread has its own WaiterEntry object.
347 347 private static WaiterEntry GetThreadWaiterEntry()
348 // Store the new stack head 348 {
349 WaiterEntry newHeadWaiterEntry = oldFirstWaiterEntry._nextWaiterEntry; 349 if (null == CurrentWaiterEntry)
350 350 {
351 // Update the old stack head list links and decrement the number 351 CurrentWaiterEntry = new WaiterEntry();
352 // waiters. 352 }
353 RemoveWaiter(oldFirstWaiterEntry, true); 353 CurrentWaiterEntry.Reset();
354 354 return CurrentWaiterEntry;
355 // Update the new stack head 355 }
356 _headWaiterEntry._nextWaiterEntry = newHeadWaiterEntry; 356
357 if (null != newHeadWaiterEntry) 357 #region Waiters stack methods
358 { 358
359 newHeadWaiterEntry._prevWaiterEntry = _headWaiterEntry; 359 /// <summary>
360 } 360 /// Push a new waiter into the waiter's stack
361 361 /// </summary>
362 // Return the old stack head 362 /// <param name="newWaiterEntry">A waiter to put in the stack</param>
363 return oldFirstWaiterEntry; 363 public void PushWaiter(WaiterEntry newWaiterEntry)
364 } 364 {
365 365 // Remove the waiter if it is already in the stack and
366 /// <summary> 366 // update waiter's count as needed
367 /// Remove a waiter from the stack 367 RemoveWaiter(newWaiterEntry, false);
368 /// </summary> 368
369 /// <param name="waiterEntry">A waiter entry to remove</param> 369 // If the stack is empty then newWaiterEntry is the new head of the stack
370 /// <param name="popDecrement">If true the waiter count is always decremented</param> 370 if (null == _headWaiterEntry._nextWaiterEntry)
371 private void RemoveWaiter(WaiterEntry waiterEntry, bool popDecrement) 371 {
372 { 372 _headWaiterEntry._nextWaiterEntry = newWaiterEntry;
373 // Store the prev entry in the list 373 newWaiterEntry._prevWaiterEntry = _headWaiterEntry;
374 WaiterEntry prevWaiterEntry = waiterEntry._prevWaiterEntry; 374
375 375 }
376 // Store the next entry in the list 376 // If the stack is not empty then put newWaiterEntry as the new head
377 WaiterEntry nextWaiterEntry = waiterEntry._nextWaiterEntry; 377 // of the stack.
378 378 else
379 // A flag to indicate if we need to decrement the waiters count. 379 {
380 // If we got here from PopWaiter then we must decrement. 380 // Save the old first waiter entry
381 // If we got here from PushWaiter then we decrement only if 381 WaiterEntry oldFirstWaiterEntry = _headWaiterEntry._nextWaiterEntry;
382 // the waiter was already in the stack. 382
383 bool decrementCounter = popDecrement; 383 // Update the links
384 384 _headWaiterEntry._nextWaiterEntry = newWaiterEntry;
385 // Null the waiter's entry links 385 newWaiterEntry._nextWaiterEntry = oldFirstWaiterEntry;
386 waiterEntry._prevWaiterEntry = null; 386 newWaiterEntry._prevWaiterEntry = _headWaiterEntry;
387 waiterEntry._nextWaiterEntry = null; 387 oldFirstWaiterEntry._prevWaiterEntry = newWaiterEntry;
388 388 }
389 // If the waiter entry had a prev link then update it. 389
390 // It also means that the waiter is already in the list and we 390 // Increment the number of waiters
391 // need to decrement the waiters count. 391 ++_waitersCount;
392 if (null != prevWaiterEntry) 392 }
393 { 393
394 prevWaiterEntry._nextWaiterEntry = nextWaiterEntry; 394 /// <summary>
395 decrementCounter = true; 395 /// Pop a waiter from the waiter's stack
396 } 396 /// </summary>
397 397 /// <returns>Returns the first waiter in the stack</returns>
398 // If the waiter entry had a next link then update it. 398 private WaiterEntry PopWaiter()
399 // It also means that the waiter is already in the list and we 399 {
400 // need to decrement the waiters count. 400 // Store the current stack head
401 if (null != nextWaiterEntry) 401 WaiterEntry oldFirstWaiterEntry = _headWaiterEntry._nextWaiterEntry;
402 { 402
403 nextWaiterEntry._prevWaiterEntry = prevWaiterEntry; 403 // Store the new stack head
404 decrementCounter = true; 404 WaiterEntry newHeadWaiterEntry = oldFirstWaiterEntry._nextWaiterEntry;
405 } 405
406 406 // Update the old stack head list links and decrement the number
407 // Decrement the waiters count if needed 407 // waiters.
408 if (decrementCounter) 408 RemoveWaiter(oldFirstWaiterEntry, true);
409 { 409
410 --_waitersCount; 410 // Update the new stack head
411 } 411 _headWaiterEntry._nextWaiterEntry = newHeadWaiterEntry;
412 } 412 if (null != newHeadWaiterEntry)
413 413 {
414 #endregion 414 newHeadWaiterEntry._prevWaiterEntry = _headWaiterEntry;
415 415 }
416 #endregion 416
417 417 // Return the old stack head
418 #region WaiterEntry class 418 return oldFirstWaiterEntry;
419 419 }
420 // A waiter entry in the _waiters queue. 420
421 public class WaiterEntry : IDisposable 421 /// <summary>
422 { 422 /// Remove a waiter from the stack
423 #region Member variables 423 /// </summary>
424 424 /// <param name="waiterEntry">A waiter entry to remove</param>
425 /// <summary> 425 /// <param name="popDecrement">If true the waiter count is always decremented</param>
426 /// Event to signal the waiter that it got the work item. 426 private void RemoveWaiter(WaiterEntry waiterEntry, bool popDecrement)
427 /// </summary> 427 {
428 private AutoResetEvent _waitHandle = new AutoResetEvent(false); 428 // Store the prev entry in the list
429 429 WaiterEntry prevWaiterEntry = waiterEntry._prevWaiterEntry;
430 /// <summary> 430
431 /// Flag to know if this waiter already quited from the queue 431 // Store the next entry in the list
432 /// because of a timeout. 432 WaiterEntry nextWaiterEntry = waiterEntry._nextWaiterEntry;
433 /// </summary> 433
434 private bool _isTimedout = false; 434 // A flag to indicate if we need to decrement the waiters count.
435 435 // If we got here from PopWaiter then we must decrement.
436 /// <summary> 436 // If we got here from PushWaiter then we decrement only if
437 /// Flag to know if the waiter was signaled and got a work item. 437 // the waiter was already in the stack.
438 /// </summary> 438 bool decrementCounter = popDecrement;
439 private bool _isSignaled = false; 439
440 440 // Null the waiter's entry links
441 /// <summary> 441 waiterEntry._prevWaiterEntry = null;
442 /// A work item that passed directly to the waiter withou going 442 waiterEntry._nextWaiterEntry = null;
443 /// through the queue 443
444 /// </summary> 444 // If the waiter entry had a prev link then update it.
445 private WorkItem _workItem = null; 445 // It also means that the waiter is already in the list and we
446 446 // need to decrement the waiters count.
447 private bool _isDisposed = false; 447 if (null != prevWaiterEntry)
448 448 {
449 // Linked list members 449 prevWaiterEntry._nextWaiterEntry = nextWaiterEntry;
450 internal WaiterEntry _nextWaiterEntry = null; 450 decrementCounter = true;
451 internal WaiterEntry _prevWaiterEntry = null; 451 }
452 452
453 #endregion 453 // If the waiter entry had a next link then update it.
454 454 // It also means that the waiter is already in the list and we
455 #region Construction 455 // need to decrement the waiters count.
456 456 if (null != nextWaiterEntry)
457 public WaiterEntry() 457 {
458 { 458 nextWaiterEntry._prevWaiterEntry = prevWaiterEntry;
459 Reset(); 459 decrementCounter = true;
460 } 460 }
461 461
462 #endregion 462 // Decrement the waiters count if needed
463 463 if (decrementCounter)
464 #region Public methods 464 {
465 465 --_waitersCount;
466 public WaitHandle WaitHandle 466 }
467 { 467 }
468 get { return _waitHandle; } 468
469 } 469 #endregion
470 470
471 public WorkItem WorkItem 471 #endregion
472 { 472
473 get 473 #region WaiterEntry class
474 { 474
475 lock(this) 475 // A waiter entry in the _waiters queue.
476 { 476 public sealed class WaiterEntry : IDisposable
477 return _workItem; 477 {
478 } 478 #region Member variables
479 } 479
480 } 480 /// <summary>
481 481 /// Event to signal the waiter that it got the work item.
482 /// <summary> 482 /// </summary>
483 /// Signal the waiter that it got a work item. 483 //private AutoResetEvent _waitHandle = new AutoResetEvent(false);
484 /// </summary> 484 private AutoResetEvent _waitHandle = EventWaitHandleFactory.CreateAutoResetEvent();
485 /// <returns>Return true on success</returns> 485
486 /// The method fails if Timeout() preceded its call 486 /// <summary>
487 public bool Signal(WorkItem workItem) 487 /// Flag to know if this waiter already quited from the queue
488 { 488 /// because of a timeout.
489 lock(this) 489 /// </summary>
490 { 490 private bool _isTimedout = false;
491 if (!_isTimedout) 491
492 { 492 /// <summary>
493 _workItem = workItem; 493 /// Flag to know if the waiter was signaled and got a work item.
494 _isSignaled = true; 494 /// </summary>
495 _waitHandle.Set(); 495 private bool _isSignaled = false;
496 return true; 496
497 } 497 /// <summary>
498 } 498 /// A work item that passed directly to the waiter withou going
499 return false; 499 /// through the queue
500 } 500 /// </summary>
501 501 private WorkItem _workItem = null;
502 /// <summary> 502
503 /// Mark the wait entry that it has been timed out 503 private bool _isDisposed = false;
504 /// </summary> 504
505 /// <returns>Return true on success</returns> 505 // Linked list members
506 /// The method fails if Signal() preceded its call 506 internal WaiterEntry _nextWaiterEntry = null;
507 public bool Timeout() 507 internal WaiterEntry _prevWaiterEntry = null;
508 { 508
509 lock(this) 509 #endregion
510 { 510
511 // Time out can happen only if the waiter wasn't marked as 511 #region Construction
512 // signaled 512
513 if (!_isSignaled) 513 public WaiterEntry()
514 { 514 {
515 // We don't remove the waiter from the queue, the DequeueWorkItem 515 Reset();
516 // method skips _waiters that were timed out. 516 }
517 _isTimedout = true; 517
518 return true; 518 #endregion
519 } 519
520 } 520 #region Public methods
521 return false; 521
522 } 522 public WaitHandle WaitHandle
523 523 {
524 /// <summary> 524 get { return _waitHandle; }
525 /// Reset the wait entry so it can be used again 525 }
526 /// </summary> 526
527 public void Reset() 527 public WorkItem WorkItem
528 { 528 {
529 _workItem = null; 529 get
530 _isTimedout = false; 530 {
531 _isSignaled = false; 531 return _workItem;
532 _waitHandle.Reset(); 532 }
533 } 533 }
534 534
535 /// <summary> 535 /// <summary>
536 /// Free resources 536 /// Signal the waiter that it got a work item.
537 /// </summary> 537 /// </summary>
538 public void Close() 538 /// <returns>Return true on success</returns>
539 { 539 /// The method fails if Timeout() preceded its call
540 if (null != _waitHandle) 540 public bool Signal(WorkItem workItem)
541 { 541 {
542 _waitHandle.Close(); 542 lock(this)
543 _waitHandle = null; 543 {
544 } 544 if (!_isTimedout)
545 } 545 {
546 546 _workItem = workItem;
547 #endregion 547 _isSignaled = true;
548 548 _waitHandle.Set();
549 #region IDisposable Members 549 return true;
550 550 }
551 public void Dispose() 551 }
552 { 552 return false;
553 if (!_isDisposed) 553 }
554 { 554
555 Close(); 555 /// <summary>
556 _isDisposed = true; 556 /// Mark the wait entry that it has been timed out
557 } 557 /// </summary>
558 } 558 /// <returns>Return true on success</returns>
559 559 /// The method fails if Signal() preceded its call
560 ~WaiterEntry() 560 public bool Timeout()
561 { 561 {
562 Dispose(); 562 lock(this)
563 } 563 {
564 564 // Time out can happen only if the waiter wasn't marked as
565 #endregion 565 // signaled
566 } 566 if (!_isSignaled)
567 567 {
568 #endregion 568 // We don't remove the waiter from the queue, the DequeueWorkItem
569 569 // method skips _waiters that were timed out.
570 #region IDisposable Members 570 _isTimedout = true;
571 571 return true;
572 public void Dispose() 572 }
573 { 573 }
574 if (!_isDisposed) 574 return false;
575 { 575 }
576 Cleanup(); 576
577 _isDisposed = true; 577 /// <summary>
578 GC.SuppressFinalize(this); 578 /// Reset the wait entry so it can be used again
579 } 579 /// </summary>
580 } 580 public void Reset()
581 581 {
582 ~WorkItemsQueue() 582 _workItem = null;
583 { 583 _isTimedout = false;
584 Cleanup(); 584 _isSignaled = false;
585 } 585 _waitHandle.Reset();
586 586 }
587 private void ValidateNotDisposed() 587
588 { 588 /// <summary>
589 if(_isDisposed) 589 /// Free resources
590 { 590 /// </summary>
591 throw new ObjectDisposedException(GetType().ToString(), "The SmartThreadPool has been shutdown"); 591 public void Close()
592 } 592 {
593 } 593 if (null != _waitHandle)
594 594 {
595 #endregion 595 _waitHandle.Close();
596 } 596 _waitHandle = null;
597 597 }
598 #endregion 598 }
599} 599
600 600 #endregion
601
602 #region IDisposable Members
603
604 public void Dispose()
605 {
606 lock (this)
607 {
608 if (!_isDisposed)
609 {
610 Close();
611 }
612 _isDisposed = true;
613 }
614 }
615
616 #endregion
617 }
618
619 #endregion
620
621 #region IDisposable Members
622
623 public void Dispose()
624 {
625 if (!_isDisposed)
626 {
627 Cleanup();
628 }
629 _isDisposed = true;
630 }
631
632 private void ValidateNotDisposed()
633 {
634 if(_isDisposed)
635 {
636 throw new ObjectDisposedException(GetType().ToString(), "The SmartThreadPool has been shutdown");
637 }
638 }
639
640 #endregion
641 }
642
643 #endregion
644}
645
generated by cgit v1.1 at 2024-12-16 10:42:34 +0000