Mercurial > pub > ImplabNet
annotate Implab/Parallels/DispatchPool.cs @ 21:6a56df4ec59e promises
DispatchPool works again, but performance is poor in some cases
| author | cin |
|---|---|
| date | Tue, 12 Nov 2013 19:52:10 +0400 |
| parents | 1c3b3d518480 |
| children | 5a35900264f5 |
| rev | line source |
|---|---|
| 15 | 1 using System; |
| 2 using System.Collections.Generic; | |
| 3 using System.Linq; | |
| 4 using System.Text; | |
| 5 using System.Threading; | |
| 6 using System.Diagnostics; | |
| 7 | |
| 8 namespace Implab.Parallels { | |
| 9 public abstract class DispatchPool<TUnit> : IDisposable { | |
| 10 readonly int m_minThreads; | |
| 11 readonly int m_maxThreads; | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
12 |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
13 int m_createdThreads = 0; // the current size of the pool |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
14 int m_activeThreads = 0; // the count of threads which are active |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
15 int m_sleepingThreads = 0; // the count of currently inactive threads |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
16 int m_maxRunningThreads = 0; // the meximum reached size of the pool |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
17 int m_exitRequired = 0; // the pool is going to shutdown, all unused workers are released |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
18 int m_releaseTimeout = 100; // the timeout while the working thread will wait for the new tasks before exit |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
19 int m_wakeEvents = 0; // the count of wake events |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
20 |
| 15 | 21 AutoResetEvent m_hasTasks = new AutoResetEvent(false); |
| 22 | |
| 23 protected DispatchPool(int min, int max) { | |
| 24 if (min < 0) | |
| 25 throw new ArgumentOutOfRangeException("min"); | |
| 26 if (max <= 0) | |
| 27 throw new ArgumentOutOfRangeException("max"); | |
| 28 | |
| 29 if (min > max) | |
| 30 min = max; | |
| 31 m_minThreads = min; | |
| 32 m_maxThreads = max; | |
| 33 } | |
| 34 | |
| 35 protected DispatchPool(int threads) | |
| 36 : this(threads, threads) { | |
| 37 } | |
| 38 | |
| 39 protected DispatchPool() { | |
| 40 int maxThreads, maxCP; | |
| 41 ThreadPool.GetMaxThreads(out maxThreads, out maxCP); | |
| 42 | |
| 43 m_minThreads = 0; | |
| 44 m_maxThreads = maxThreads; | |
| 45 } | |
| 46 | |
| 47 protected void InitPool() { | |
| 48 for (int i = 0; i < m_minThreads; i++) | |
| 49 StartWorker(); | |
| 50 } | |
| 51 | |
| 20 | 52 public int PoolSize { |
| 15 | 53 get { |
| 20 | 54 return m_createdThreads; |
| 55 } | |
| 56 } | |
| 57 | |
| 58 public int ActiveThreads { | |
| 59 get { | |
| 60 return m_activeThreads; | |
| 15 | 61 } |
| 62 } | |
| 63 | |
| 64 public int MaxRunningThreads { | |
| 65 get { | |
| 66 return m_maxRunningThreads; | |
| 67 } | |
| 68 } | |
| 69 | |
| 70 protected bool IsDisposed { | |
| 71 get { | |
| 72 return m_exitRequired != 0; | |
| 73 } | |
| 74 } | |
| 75 | |
| 17 | 76 protected abstract bool TryDequeue(out TUnit unit); |
| 77 | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
78 #region thread execution traits |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
79 int SignalThread() { |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
80 var signals = Interlocked.Increment(ref m_wakeEvents); |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
81 if(signals == 1) |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
82 m_hasTasks.Set(); |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
83 return signals; |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
84 } |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
85 |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
86 bool Sleep(int timeout) { |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
87 Interlocked.Increment(ref m_sleepingThreads); |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
88 if (m_hasTasks.WaitOne(timeout)) { |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
89 // this is autoreset event, only one thread can run this block simultaneously |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
90 var sleeping = Interlocked.Decrement(ref m_sleepingThreads); |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
91 if (Interlocked.Decrement(ref m_wakeEvents) > 0) |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
92 m_hasTasks.Set(); // wake next worker |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
93 |
| 17 | 94 return true; |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
95 } else { |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
96 Interlocked.Decrement(ref m_sleepingThreads); |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
97 return false; |
| 20 | 98 } |
| 17 | 99 } |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
100 #endregion |
| 17 | 101 |
| 102 /// <summary> | |
| 103 /// Запускает либо новый поток, если раньше не было ни одного потока, либо устанавливает событие пробуждение одного спящего потока | |
| 104 /// </summary> | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
105 protected void GrowPool() { |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
106 if (m_exitRequired != 0) |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
107 return; |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
108 if (m_sleepingThreads > m_wakeEvents) { |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
109 // all sleeping threads may gone |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
110 SignalThread(); // wake a sleeping thread; |
| 17 | 111 |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
112 // we can't check whether signal has been processed |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
113 // anyway it may take some time for the thread to start |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
114 // we will ensure that at least one thread is running |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
115 |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
116 if (AllocateThreadSlot(1)) { |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
117 // if there were no threads in the pool |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
118 var worker = new Thread(this.Worker); |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
119 worker.IsBackground = true; |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
120 worker.Start(); |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
121 } |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
122 } else { |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
123 // if there is no sleeping threads in the pool |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
124 StartWorker(); |
| 17 | 125 } |
| 126 } | |
| 127 | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
128 private bool Suspend() { |
| 20 | 129 //no tasks left, exit if the thread is no longer needed |
| 130 bool last; | |
| 131 bool requestExit; | |
| 132 | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
133 |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
134 |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
135 // if threads have a timeout before releasing |
| 20 | 136 if (m_releaseTimeout > 0) |
| 137 requestExit = !Sleep(m_releaseTimeout); | |
| 138 else | |
| 139 requestExit = true; | |
| 140 | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
141 if (!requestExit) |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
142 return true; |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
143 |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
144 // release unsused thread |
| 20 | 145 if (requestExit && ReleaseThreadSlot(out last)) { |
| 146 // in case at the moment the last thread was being released | |
| 147 // a new task was added to the queue, we need to try | |
| 148 // to revoke the thread to avoid the situation when the task is left unprocessed | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
149 if (last && Sleep(0)) { // Sleep(0) will fetch pending task or will return false |
| 20 | 150 if (AllocateThreadSlot(1)) |
| 151 return true; // spin again... | |
| 152 else | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
153 SignalThread(); // since Sleep(0) has fetched the signal we neet to reschedule it |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
154 |
| 20 | 155 } |
| 156 | |
| 157 return false; | |
| 158 } | |
| 159 | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
160 // wait till infinity |
| 20 | 161 Sleep(-1); |
| 162 | |
| 163 return true; | |
| 17 | 164 } |
| 165 | |
| 166 #region thread slots traits | |
| 167 | |
| 168 bool AllocateThreadSlot() { | |
| 16 | 169 int current; |
| 15 | 170 // use spins to allocate slot for the new thread |
| 171 do { | |
| 20 | 172 current = m_createdThreads; |
| 15 | 173 if (current >= m_maxThreads || m_exitRequired != 0) |
| 174 // no more slots left or the pool has been disposed | |
| 175 return false; | |
| 20 | 176 } while (current != Interlocked.CompareExchange(ref m_createdThreads, current + 1, current)); |
| 15 | 177 |
| 17 | 178 UpdateMaxThreads(current + 1); |
| 179 | |
| 180 return true; | |
| 181 } | |
| 15 | 182 |
| 17 | 183 bool AllocateThreadSlot(int desired) { |
| 20 | 184 if (desired - 1 != Interlocked.CompareExchange(ref m_createdThreads, desired, desired - 1)) |
| 17 | 185 return false; |
| 186 | |
| 187 UpdateMaxThreads(desired); | |
| 15 | 188 |
| 17 | 189 return true; |
| 190 } | |
| 191 | |
| 192 bool ReleaseThreadSlot(out bool last) { | |
| 193 last = false; | |
| 194 int current; | |
| 195 // use spins to release slot for the new thread | |
| 196 do { | |
| 20 | 197 current = m_createdThreads; |
| 17 | 198 if (current <= m_minThreads && m_exitRequired == 0) |
| 199 // the thread is reserved | |
| 200 return false; | |
| 20 | 201 } while (current != Interlocked.CompareExchange(ref m_createdThreads, current - 1, current)); |
| 17 | 202 |
| 203 last = (current == 1); | |
| 15 | 204 |
| 205 return true; | |
| 206 } | |
| 207 | |
| 17 | 208 /// <summary> |
| 209 /// releases thread slot unconditionally, used during cleanup | |
| 210 /// </summary> | |
| 211 /// <returns>true - no more threads left</returns> | |
| 212 bool ReleaseThreadSlotAnyway() { | |
| 20 | 213 var left = Interlocked.Decrement(ref m_createdThreads); |
| 17 | 214 return left == 0; |
| 15 | 215 } |
| 216 | |
| 17 | 217 void UpdateMaxThreads(int count) { |
| 218 int max; | |
| 16 | 219 do { |
| 17 | 220 max = m_maxRunningThreads; |
| 221 if (max >= count) | |
| 222 break; | |
| 223 } while(max != Interlocked.CompareExchange(ref m_maxRunningThreads, count, max)); | |
| 16 | 224 } |
| 225 | |
| 17 | 226 #endregion |
| 227 | |
| 228 bool StartWorker() { | |
| 229 if (AllocateThreadSlot()) { | |
| 230 // slot successfully allocated | |
| 231 var worker = new Thread(this.Worker); | |
| 232 worker.IsBackground = true; | |
| 233 worker.Start(); | |
| 234 | |
| 235 return true; | |
| 236 } else { | |
| 237 return false; | |
| 238 } | |
| 16 | 239 } |
| 240 | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
241 protected abstract void InvokeUnit(TUnit unit); |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
242 |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
243 void Worker() { |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
244 TUnit unit; |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
245 Interlocked.Increment(ref m_activeThreads); |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
246 Sleep(0); // remove wake request if the new thread is started |
| 15 | 247 do { |
| 248 // exit if requested | |
| 249 if (m_exitRequired != 0) { | |
| 250 // release the thread slot | |
| 20 | 251 Interlocked.Decrement(ref m_activeThreads); |
| 17 | 252 if (ReleaseThreadSlotAnyway()) // it was the last worker |
| 15 | 253 m_hasTasks.Dispose(); |
| 254 else | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
255 SignalThread(); // wake next worker |
| 15 | 256 unit = default(TUnit); |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
257 break; |
| 15 | 258 } |
| 259 | |
| 260 // fetch task | |
| 261 if (TryDequeue(out unit)) { | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
262 InvokeUnit(unit); |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
263 continue; |
| 15 | 264 } |
| 265 | |
| 20 | 266 Interlocked.Decrement(ref m_activeThreads); |
| 15 | 267 |
| 16 | 268 // entering suspend state |
| 269 // keep this thread and wait | |
| 20 | 270 if (!Suspend()) |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
271 break; |
| 20 | 272 |
| 273 Interlocked.Increment(ref m_activeThreads); | |
| 15 | 274 } while (true); |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
275 |
| 15 | 276 } |
| 277 | |
| 278 protected virtual void Dispose(bool disposing) { | |
| 279 if (disposing) { | |
| 280 if (m_exitRequired == 0) { | |
| 281 if (Interlocked.CompareExchange(ref m_exitRequired, 1, 0) != 0) | |
| 282 return; | |
| 283 | |
| 284 // wake sleeping threads | |
|
21
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
285 if (m_createdThreads > 0) |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
286 SignalThread(); |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
287 else |
|
6a56df4ec59e
DispatchPool works again, but performance is poor in some cases
cin
parents:
20
diff
changeset
|
288 m_hasTasks.Dispose(); |
| 15 | 289 GC.SuppressFinalize(this); |
| 290 } | |
| 291 } | |
| 292 } | |
| 293 | |
| 294 public void Dispose() { | |
| 295 Dispose(true); | |
| 296 } | |
| 297 | |
| 298 ~DispatchPool() { | |
| 299 Dispose(false); | |
| 300 } | |
| 301 } | |
| 302 } |