Mercurial > pub > ImplabNet
comparison Implab/Parallels/DispatchPool.cs @ 81:2c5631b43c7d v2
dispatch pool rewritten
author | cin |
---|---|
date | Fri, 26 Sep 2014 20:44:01 +0400 |
parents | 4f20870d0816 |
children | ce0171cacec4 |
comparison
equal
deleted
inserted
replaced
80:4f20870d0816 | 81:2c5631b43c7d |
---|---|
5 using System.Threading; | 5 using System.Threading; |
6 using System.Diagnostics; | 6 using System.Diagnostics; |
7 | 7 |
8 namespace Implab.Parallels { | 8 namespace Implab.Parallels { |
9 public abstract class DispatchPool<TUnit> : IDisposable { | 9 public abstract class DispatchPool<TUnit> : IDisposable { |
10 readonly int m_minThreads; | 10 readonly int m_minThreadsLimit; |
11 readonly int m_maxThreads; | 11 readonly int m_maxThreadsLimit; |
12 readonly int m_releaseTimeout = 100; // the timeout while the working thread will wait for the new tasks before exit | 12 readonly int m_releaseTimeout = 1000; // the timeout while the working thread will wait for the new tasks before exit |
13 | 13 |
14 int m_createdThreads = 0; // the current size of the pool | 14 int m_threads = 0; // the current size of the pool |
15 int m_activeThreads = 0; // the count of threads which are active | |
16 int m_sleepingThreads = 0; // the count of currently inactive threads | |
17 int m_maxRunningThreads = 0; // the meximum reached size of the pool | 15 int m_maxRunningThreads = 0; // the meximum reached size of the pool |
18 int m_exitRequired = 0; // the pool is going to shutdown, all unused workers are released | 16 int m_exit = 0; // the pool is going to shutdown, all unused workers are released |
19 | 17 |
20 int m_wakeEvents = 0; // the count of wake events | 18 readonly object m_signal = new object(); // used to pulse waiting threads |
21 | |
22 readonly object m_signalLocker = new object(); | |
23 | 19 |
24 protected DispatchPool(int min, int max) { | 20 protected DispatchPool(int min, int max) { |
25 if (min < 0) | 21 if (min < 0) |
26 throw new ArgumentOutOfRangeException("min"); | 22 throw new ArgumentOutOfRangeException("min"); |
27 if (max <= 0) | 23 if (max <= 0) |
28 throw new ArgumentOutOfRangeException("max"); | 24 throw new ArgumentOutOfRangeException("max"); |
29 | 25 |
30 if (min > max) | 26 if (min > max) |
31 min = max; | 27 min = max; |
32 m_minThreads = min; | 28 m_minThreadsLimit = min; |
33 m_maxThreads = max; | 29 m_maxThreadsLimit = max; |
34 } | 30 } |
35 | 31 |
36 protected DispatchPool(int threads) | 32 protected DispatchPool(int threads) |
37 : this(threads, threads) { | 33 : this(threads, threads) { |
38 } | 34 } |
39 | 35 |
40 protected DispatchPool() { | 36 protected DispatchPool() { |
41 int maxThreads, maxCP; | 37 int maxThreads, maxCP; |
42 ThreadPool.GetMaxThreads(out maxThreads, out maxCP); | 38 ThreadPool.GetMaxThreads(out maxThreads, out maxCP); |
43 | 39 |
44 m_minThreads = 0; | 40 m_minThreadsLimit = 0; |
45 m_maxThreads = maxThreads; | 41 m_maxThreadsLimit = maxThreads; |
46 } | 42 } |
47 | 43 |
48 protected void InitPool() { | 44 protected void InitPool() { |
49 for (int i = 0; i < m_minThreads; i++) | 45 for (int i = 0; i < m_minThreadsLimit; i++) |
50 StartWorker(); | 46 StartWorker(); |
51 } | 47 } |
52 | 48 |
53 public int PoolSize { | 49 public int PoolSize { |
54 get { | 50 get { |
55 Thread.MemoryBarrier(); | 51 Thread.MemoryBarrier(); |
56 return m_createdThreads; | 52 return m_threads; |
57 } | 53 } |
58 } | 54 } |
59 | 55 |
60 public int ActiveThreads { | |
61 get { | |
62 Thread.MemoryBarrier(); | |
63 return m_activeThreads; | |
64 } | |
65 } | |
66 | |
67 public int MaxRunningThreads { | 56 public int MaxRunningThreads { |
68 get { | 57 get { |
69 Thread.MemoryBarrier(); | 58 Thread.MemoryBarrier(); |
70 return m_maxRunningThreads; | 59 return m_maxRunningThreads; |
71 } | 60 } |
72 } | 61 } |
73 | 62 |
74 protected bool IsDisposed { | 63 protected bool IsDisposed { |
75 get { | 64 get { |
76 Thread.MemoryBarrier(); | 65 Thread.MemoryBarrier(); |
77 return m_exitRequired == 1; | 66 return m_exit == 1; |
78 } | 67 } |
79 } | 68 } |
80 | 69 |
81 protected abstract bool TryDequeue(out TUnit unit); | 70 protected abstract bool TryDequeue(out TUnit unit); |
82 | 71 |
83 #region thread signaling traits | 72 private bool Dequeue(out TUnit unit, int timeout) { |
84 int SignalThread() { | 73 int ts = Environment.TickCount; |
85 var signals = Interlocked.Increment(ref m_wakeEvents); | 74 if (TryDequeue(out unit)) |
86 if(signals == 1) | 75 return true; |
87 lock(m_signalLocker) | 76 lock (m_signal) { |
88 Monitor.Pulse(m_signalLocker); | 77 while (!TryDequeue(out unit) && m_exit == 0) |
89 return signals; | 78 if(!Monitor.Wait(m_signal, Math.Max(0, ts + timeout - Environment.TickCount))) { |
90 } | 79 // timeout |
91 | 80 return false; |
92 bool FetchSignalOrWait(int timeout) { | |
93 var start = Environment.TickCount; | |
94 int signals; | |
95 Thread.MemoryBarrier(); // m_wakeEvents volatile first read | |
96 do { | |
97 signals = m_wakeEvents; | |
98 if (signals == 0) | |
99 break; | |
100 } while (Interlocked.CompareExchange(ref m_wakeEvents, signals - 1, signals) != signals); | |
101 | |
102 if (signals == 0) { | |
103 // no signal is fetched | |
104 lock(m_signalLocker) { | |
105 while(m_wakeEvents == 0) { | |
106 if (timeout != -1) | |
107 timeout = Math.Max(0, timeout - (Environment.TickCount - start)); | |
108 if(!Monitor.Wait(m_signalLocker,timeout)) | |
109 return false; // timeout | |
110 } | 81 } |
111 // m_wakeEvents > 0 | 82 // queue item or terminate |
112 if (Interlocked.Decrement(ref m_wakeEvents) > 0) //syncronized | 83 Monitor.Pulse(m_signal); |
113 Monitor.Pulse(m_signalLocker); | 84 if (m_exit == 1) |
114 | 85 return false; |
115 // signal fetched | 86 } |
116 return true; | 87 return true; |
117 } | 88 } |
118 | 89 |
119 } else { | 90 protected void SignalThread() { |
120 // signal fetched | 91 lock (m_signal) { |
121 return true; | 92 Monitor.Pulse(m_signal); |
122 } | 93 } |
123 | |
124 | |
125 } | |
126 | |
127 bool Sleep(int timeout) { | |
128 Interlocked.Increment(ref m_sleepingThreads); | |
129 if (FetchSignalOrWait(timeout)) { | |
130 Interlocked.Decrement(ref m_sleepingThreads); | |
131 return true; | |
132 } else { | |
133 Interlocked.Decrement(ref m_sleepingThreads); | |
134 return false; | |
135 } | |
136 } | |
137 #endregion | |
138 | |
139 /// <summary> | |
140 /// Запускает либо новый поток, если раньше не было ни одного потока, либо устанавливает событие пробуждение одного спящего потока | |
141 /// </summary> | |
142 protected void GrowPool() { | |
143 Thread.MemoryBarrier(); | |
144 if (m_exitRequired == 1) | |
145 return; | |
146 if (m_sleepingThreads > m_wakeEvents) { | |
147 //Console.WriteLine("Waking threads (sleeps {0}, pending {1})", m_sleepingThreads, m_wakeEvents); | |
148 | |
149 // all sleeping threads may gone | |
150 SignalThread(); // wake a sleeping thread; | |
151 | |
152 // we can't check whether signal has been processed | |
153 // anyway it may take some time for the thread to start | |
154 // we will ensure that at least one thread is running | |
155 | |
156 EnsurePoolIsAlive(); | |
157 } else { | |
158 // if there is no sleeping threads in the pool | |
159 if (!StartWorker()) { | |
160 // we haven't started a new thread, but the current can be on the way to terminate and it can't process the queue | |
161 // send it a signal to spin again | |
162 SignalThread(); | |
163 EnsurePoolIsAlive(); | |
164 } | |
165 } | |
166 } | |
167 | |
168 protected void EnsurePoolIsAlive() { | |
169 if (AllocateThreadSlot(1)) { | |
170 // if there were no threads in the pool | |
171 var worker = new Thread(this.Worker); | |
172 worker.IsBackground = true; | |
173 worker.Start(); | |
174 } | |
175 } | |
176 | |
177 protected virtual bool Suspend() { | |
178 //no tasks left, exit if the thread is no longer needed | |
179 bool last; | |
180 bool requestExit; | |
181 | |
182 // if threads have a timeout before releasing | |
183 if (m_releaseTimeout > 0) | |
184 requestExit = !Sleep(m_releaseTimeout); | |
185 else | |
186 requestExit = true; | |
187 | |
188 if (!requestExit) | |
189 return true; | |
190 | |
191 // release unsused thread | |
192 if (requestExit && ReleaseThreadSlot(out last)) { | |
193 // in case at the moment the last thread was being released | |
194 // a new task was added to the queue, we need to try | |
195 // to revoke the thread to avoid the situation when the task is left unprocessed | |
196 if (last && FetchSignalOrWait(0)) { // FetchSignalOrWait(0) will fetch pending task or will return false | |
197 SignalThread(); // since FetchSignalOrWait(0) has fetched the signal we need to reschedule it | |
198 return AllocateThreadSlot(1); // ensure that at least one thread is alive | |
199 } | |
200 | |
201 return false; | |
202 } | |
203 | |
204 // wait till infinity | |
205 Sleep(-1); | |
206 | |
207 return true; | |
208 } | 94 } |
209 | 95 |
210 #region thread slots traits | 96 #region thread slots traits |
211 | 97 |
212 bool AllocateThreadSlot() { | 98 bool AllocateThreadSlot() { |
213 int current; | 99 int current; |
214 // use spins to allocate slot for the new thread | 100 // use spins to allocate slot for the new thread |
215 do { | 101 do { |
216 current = m_createdThreads; | 102 current = m_threads; |
217 if (current >= m_maxThreads || m_exitRequired == 1) | 103 if (current >= m_maxThreadsLimit || m_exit == 1) |
218 // no more slots left or the pool has been disposed | 104 // no more slots left or the pool has been disposed |
219 return false; | 105 return false; |
220 } while (current != Interlocked.CompareExchange(ref m_createdThreads, current + 1, current)); | 106 } while (current != Interlocked.CompareExchange(ref m_threads, current + 1, current)); |
221 | 107 |
222 UpdateMaxThreads(current + 1); | 108 UpdateMaxThreads(current + 1); |
223 | 109 |
224 return true; | 110 return true; |
225 } | 111 } |
226 | 112 |
227 bool AllocateThreadSlot(int desired) { | 113 bool AllocateThreadSlot(int desired) { |
228 if (desired - 1 != Interlocked.CompareExchange(ref m_createdThreads, desired, desired - 1)) | 114 if (desired - 1 != Interlocked.CompareExchange(ref m_threads, desired, desired - 1)) |
229 return false; | 115 return false; |
230 | 116 |
231 UpdateMaxThreads(desired); | 117 UpdateMaxThreads(desired); |
232 | 118 |
233 return true; | 119 return true; |
237 last = false; | 123 last = false; |
238 int current; | 124 int current; |
239 // use spins to release slot for the new thread | 125 // use spins to release slot for the new thread |
240 Thread.MemoryBarrier(); | 126 Thread.MemoryBarrier(); |
241 do { | 127 do { |
242 current = m_createdThreads; | 128 current = m_threads; |
243 if (current <= m_minThreads && m_exitRequired == 0) | 129 if (current <= m_minThreadsLimit && m_exit == 0) |
244 // the thread is reserved | 130 // the thread is reserved |
245 return false; | 131 return false; |
246 } while (current != Interlocked.CompareExchange(ref m_createdThreads, current - 1, current)); | 132 } while (current != Interlocked.CompareExchange(ref m_threads, current - 1, current)); |
247 | 133 |
248 last = (current == 1); | 134 last = (current == 1); |
249 | 135 |
250 return true; | 136 return true; |
251 } | |
252 | |
253 /// <summary> | |
254 /// releases thread slot unconditionally, used during cleanup | |
255 /// </summary> | |
256 /// <returns>true - no more threads left</returns> | |
257 bool ReleaseThreadSlotAnyway() { | |
258 var left = Interlocked.Decrement(ref m_createdThreads); | |
259 return left == 0; | |
260 } | 137 } |
261 | 138 |
262 void UpdateMaxThreads(int count) { | 139 void UpdateMaxThreads(int count) { |
263 int max; | 140 int max; |
264 do { | 141 do { |
268 } while(max != Interlocked.CompareExchange(ref m_maxRunningThreads, count, max)); | 145 } while(max != Interlocked.CompareExchange(ref m_maxRunningThreads, count, max)); |
269 } | 146 } |
270 | 147 |
271 #endregion | 148 #endregion |
272 | 149 |
273 bool StartWorker() { | 150 protected bool StartWorker() { |
274 if (AllocateThreadSlot()) { | 151 if (AllocateThreadSlot()) { |
275 // slot successfully allocated | 152 // slot successfully allocated |
276 var worker = new Thread(this.Worker); | 153 var worker = new Thread(this.Worker); |
277 worker.IsBackground = true; | 154 worker.IsBackground = true; |
278 Interlocked.Increment(ref m_activeThreads); | |
279 worker.Start(); | 155 worker.Start(); |
280 | 156 |
281 return true; | 157 return true; |
282 } else { | 158 } else { |
283 return false; | 159 return false; |
286 | 162 |
287 protected abstract void InvokeUnit(TUnit unit); | 163 protected abstract void InvokeUnit(TUnit unit); |
288 | 164 |
289 protected virtual void Worker() { | 165 protected virtual void Worker() { |
290 TUnit unit; | 166 TUnit unit; |
291 //Console.WriteLine("{0}: Active", Thread.CurrentThread.ManagedThreadId); | 167 bool last; |
292 int count = 0;; | 168 do { |
293 Thread.MemoryBarrier(); | 169 while (Dequeue(out unit, m_releaseTimeout)) { |
294 do { | 170 InvokeUnit(unit); |
295 // exit if requested | |
296 if (m_exitRequired == 1) { | |
297 // release the thread slot | |
298 Interlocked.Decrement(ref m_activeThreads); | |
299 if (!ReleaseThreadSlotAnyway()) // it was the last worker | |
300 SignalThread(); // wake next worker | |
301 break; | |
302 } | 171 } |
303 | 172 if(!ReleaseThreadSlot(out last)) |
304 // fetch task | 173 continue; |
305 if (TryDequeue(out unit)) { | 174 // queue may be not empty |
175 if (last && TryDequeue(out unit)) { | |
306 InvokeUnit(unit); | 176 InvokeUnit(unit); |
307 count ++; | 177 if (AllocateThreadSlot(1)) |
308 continue; | 178 continue; |
179 // we can safely exit since pool is alive | |
309 } | 180 } |
310 Interlocked.Decrement(ref m_activeThreads); | 181 break; |
311 | 182 } while(true); |
312 Console.WriteLine("{0}: Suspend processed({1})", Thread.CurrentThread.ManagedThreadId,count); | 183 } |
313 // entering suspend state | 184 |
314 // keep this thread and wait | |
315 if (!Suspend()) | |
316 break; | |
317 count = 0; | |
318 //Console.WriteLine("{0}: Awake", Thread.CurrentThread.ManagedThreadId); | |
319 Interlocked.Increment(ref m_activeThreads); | |
320 } while (true); | |
321 //Console.WriteLine("{0}: Exited", Thread.CurrentThread.ManagedThreadId); | |
322 } | |
323 | 185 |
324 protected virtual void Dispose(bool disposing) { | 186 protected virtual void Dispose(bool disposing) { |
325 if (disposing) { | 187 if (disposing) { |
326 if (0 == Interlocked.CompareExchange(ref m_exitRequired, 1, 0)) { // implies memory barrier | 188 if (0 == Interlocked.CompareExchange(ref m_exit, 1, 0)) { // implies memory barrier |
327 // wake sleeping threads | 189 // wake sleeping threads |
328 if (m_createdThreads > 0) | 190 SignalThread(); |
329 SignalThread(); | |
330 GC.SuppressFinalize(this); | 191 GC.SuppressFinalize(this); |
331 } | 192 } |
332 } | 193 } |
333 } | 194 } |
334 | 195 |