4 // Copyright (c) 2008 Jérémie "Garuma" Laval
6 // Permission is hereby granted, free of charge, to any person obtaining a copy
7 // of this software and associated documentation files (the "Software"), to deal
8 // in the Software without restriction, including without limitation the rights
9 // to use, copy, modify, merge, publish, distribute, sublicense, and/or sell
10 // copies of the Software, and to permit persons to whom the Software is
11 // furnished to do so, subject to the following conditions:
13 // The above copyright notice and this permission notice shall be included in
14 // all copies or substantial portions of the Software.
16 // THE SOFTWARE IS PROVIDED "AS IS", WITHOUT WARRANTY OF ANY KIND, EXPRESS OR
17 // IMPLIED, INCLUDING BUT NOT LIMITED TO THE WARRANTIES OF MERCHANTABILITY,
18 // FITNESS FOR A PARTICULAR PURPOSE AND NONINFRINGEMENT. IN NO EVENT SHALL THE
19 // AUTHORS OR COPYRIGHT HOLDERS BE LIABLE FOR ANY CLAIM, DAMAGES OR OTHER
20 // LIABILITY, WHETHER IN AN ACTION OF CONTRACT, TORT OR OTHERWISE, ARISING FROM,
21 // OUT OF OR IN CONNECTION WITH THE SOFTWARE OR THE USE OR OTHER DEALINGS IN
27 using System.Threading;
28 using System.Collections.Concurrent;
30 namespace System.Threading.Tasks
32 internal class ThreadWorker : IDisposable
34 static Random r = new Random ();
38 readonly ThreadWorker[] others;
39 internal readonly IDequeOperations<Task> dDeque;
40 readonly IProducerConsumerCollection<Task> sharedWorkQueue;
41 readonly Action<Task> childWorkAdder;
43 // Flag to tell if workerThread is running
46 readonly bool isLocal;
47 readonly int workerLength;
48 readonly int stealingStart;
49 const int maxRetry = 5;
51 #region Sleep related fields
52 readonly SpinWait wait = new SpinWait ();
53 const int sleepThreshold = 100000;
56 Action threadInitializer;
58 public ThreadWorker (IScheduler sched, ThreadWorker[] others, IProducerConsumerCollection<Task> sharedWorkQueue,
59 int maxStackSize, ThreadPriority priority)
60 : this (sched, others, sharedWorkQueue, true, maxStackSize, priority)
64 public ThreadWorker (IScheduler sched, ThreadWorker[] others, IProducerConsumerCollection<Task> sharedWorkQueue,
65 bool createThread, int maxStackSize, ThreadPriority priority)
69 this.dDeque = new CyclicDeque<Task> ();
71 this.sharedWorkQueue = sharedWorkQueue;
72 this.workerLength = others.Length;
73 this.isLocal = !createThread;
75 this.childWorkAdder = delegate (Task t) {
76 dDeque.PushBottom (t);
80 // Find the stealing start index randomly (then the traversal
81 // will be done in Round-Robin fashion)
83 this.stealingStart = r.Next(0, workerLength);
84 } while (others[stealingStart] == this);
86 InitializeUnderlyingThread (maxStackSize, priority);
89 void InitializeUnderlyingThread (int maxStackSize, ThreadPriority priority)
91 threadInitializer = delegate {
92 // Special case of the participant ThreadWorker
94 this.workerThread = Thread.CurrentThread;
98 this.workerThread = (maxStackSize == 0) ? new Thread (WorkerMethodWrapper) :
99 new Thread (WorkerMethodWrapper, maxStackSize);
101 this.workerThread.IsBackground = true;
102 this.workerThread.Priority = priority;
104 threadInitializer ();
107 public void Dispose ()
110 if (!isLocal && workerThread.ThreadState != ThreadState.Stopped)
111 workerThread.Abort ();
116 // If the thread was stopped then set it in use and restart it
117 int result = Interlocked.Exchange (ref started, 1);
121 if (this.workerThread.ThreadState != ThreadState.Unstarted) {
122 threadInitializer ();
124 workerThread.Start ();
130 // Set the flag to stop so that the while in the thread will stop
131 // doing its infinite loop.
135 // This is the actual method called in the Thread
136 void WorkerMethodWrapper ()
141 while (started == 1) {
144 result = WorkerMethod ();
145 } catch (Exception e) {
146 Console.WriteLine (e.ToString ());
149 // Wait a little and if the Thread has been more sleeping than working shut it down
153 if (sleepTime++ > sleepThreshold)
160 // Main method, used to do all the logic of retrieving, processing and stealing work.
164 bool hasStolenFromOther;
166 hasStolenFromOther = false;
170 // We fill up our work deque concurrently with other ThreadWorker
171 while (sharedWorkQueue.Count > 0) {
172 while (sharedWorkQueue.TryTake (out value)) {
173 dDeque.PushBottom (value);
176 // Now we process our work
177 while (dDeque.PopBottom (out value) == PopResult.Succeed) {
179 value.Execute (childWorkAdder);
185 // When we have finished, steal from other worker
188 // Repeat the operation a little so that we can let other things process.
189 for (int j = 0; j < maxRetry; j++) {
190 // Start stealing with the ThreadWorker at our right to minimize contention
191 for (int it = stealingStart; it < stealingStart + workerLength; it++) {
192 int i = it % workerLength;
193 if ((other = others [i]) == null || other == this)
196 // Maybe make this steal more than one item at a time, see TODO.
197 if (other.dDeque.PopTop (out value) == PopResult.Succeed) {
198 hasStolenFromOther = true;
200 value.Execute (childWorkAdder);
206 } while (sharedWorkQueue.Count > 0 || hasStolenFromOther);
211 // Almost same as above but with an added predicate and treating one item at a time.
212 // It's used by Scheduler Participate(...) method for special waiting case like
213 // Task.WaitAll(someTasks) or Task.WaitAny(someTasks)
214 // Predicate should be really fast and not blocking as it is called a good deal of time
215 // Also, the method skip tasks that are LongRunning to avoid blocking (Task are not LongRunning by default)
216 public static void WorkerMethod (Func<bool> predicate, IProducerConsumerCollection<Task> sharedWorkQueue,
217 ThreadWorker[] others)
219 while (!predicate ()) {
222 // Dequeue only one item as we have restriction
223 if (sharedWorkQueue.TryTake (out value)) {
225 if (CheckTaskFitness (value))
226 value.Execute (null);
228 sharedWorkQueue.TryAdd (value);
232 // First check to see if we comply to predicate
237 // Try to complete other work by stealing since our desired tasks may be in other worker
239 for (int i = 0; i < others.Length; i++) {
240 if ((other = others [i]) == null)
243 if (other.dDeque.PopTop (out value) == PopResult.Succeed) {
245 if (CheckTaskFitness (value))
246 value.Execute (null);
248 sharedWorkQueue.TryAdd (value);
259 static bool CheckTaskFitness (Task t)
261 return (t.CreationOptions | TaskCreationOptions.LongRunning) > 0;
264 public bool Finished {
270 public bool IsLocal {
278 return workerThread.ManagedThreadId;
282 public bool Equals (ThreadWorker other)
284 return (other == null) ? false : object.ReferenceEquals (this.dDeque, other.dDeque);
287 public override bool Equals (object obj)
289 ThreadWorker temp = obj as ThreadWorker;
290 return temp == null ? false : Equals (temp);
293 public override int GetHashCode ()
295 return workerThread.ManagedThreadId.GetHashCode ();