##// END OF EJS Templates
implab » ImplabNet
RSS
dispatch pool rewritten
cin -
r81:2c5631b43c7d v2
parent child
Show More
Show file before | Show file after | Hide comments
@@ -148,7 +148,7 namespace Implab.Test {
148
148
149 [TestMethod]
149 [TestMethod]
150 public void WorkerPoolSizeTest() {
150 public void WorkerPoolSizeTest() {
151 var pool = new WorkerPool(5, 10, 0);
151 var pool = new WorkerPool(5, 10, 1);
152
152
153 Assert.AreEqual(5, pool.PoolSize);
153 Assert.AreEqual(5, pool.PoolSize);
154
154
@@ -291,7 +291,7 namespace Implab.Test {
291 [TestMethod]
291 [TestMethod]
292 public void ChainedMapTest() {
292 public void ChainedMapTest() {
293
293
294 using (var pool = new WorkerPool(0,10,100)) {
294 using (var pool = new WorkerPool(0,10,1)) {
295 const int count = 10000;
295 const int count = 10000;
296
296
297 var args = new double[count];
297 var args = new double[count];
Show file before | Show file after | Hide comments
@@ -7,19 +7,15 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)
@@ -29,8 +25,8 namespace Implab.Parallels {
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)
@@ -41,26 +37,19 namespace Implab.Parallels {
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 }
58 }
59
60 public int ActiveThreads {
61 get {
62 Thread.MemoryBarrier();
63 return m_activeThreads;
64 }
53 }
65 }
54 }
66
55
@@ -74,150 +63,47 namespace Implab.Parallels {
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))) {
79 // timeout
80 return false;
90 }
81 }
91
82 // queue item or terminate
92 bool FetchSignalOrWait(int timeout) {
83 Monitor.Pulse(m_signal);
93 var start = Environment.TickCount;
84 if (m_exit == 1)
94 int signals;
85 return false;
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 }
86 }
111 // m_wakeEvents > 0
112 if (Interlocked.Decrement(ref m_wakeEvents) > 0) //syncronized
113 Monitor.Pulse(m_signalLocker);
114
115 // signal fetched
116 return true;
117 }
118
119 } else {
120 // signal fetched
121 return true;
87 return true;
122 }
88 }
123
89
124
90 protected void SignalThread() {
125 }
91 lock (m_signal) {
126
92 Monitor.Pulse(m_signal);
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 }
93 }
166 }
94 }
167
95
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 }
209
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
@@ -225,7 +111,7 namespace Implab.Parallels {
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);
@@ -239,26 +125,17 namespace Implab.Parallels {
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 }
137 }
252
138
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 }
261
262 void UpdateMaxThreads(int count) {
139 void UpdateMaxThreads(int count) {
263 int max;
140 int max;
264 do {
141 do {
@@ -270,12 +147,11 namespace Implab.Parallels {
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;
@@ -288,44 +164,29 namespace Implab.Parallels {
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;;
293 Thread.MemoryBarrier();
294 do {
168 do {
295 // exit if requested
169 while (Dequeue(out unit, m_releaseTimeout)) {
296 if (m_exitRequired == 1) {
170 InvokeUnit(unit);
297 // release the thread slot
171 }
298 Interlocked.Decrement(ref m_activeThreads);
172 if(!ReleaseThreadSlot(out last))
299 if (!ReleaseThreadSlotAnyway()) // it was the last worker
173 continue;
300 SignalThread(); // wake next worker
174 // queue may be not empty
175 if (last && TryDequeue(out unit)) {
176 InvokeUnit(unit);
177 if (AllocateThreadSlot(1))
178 continue;
179 // we can safely exit since pool is alive
180 }
301 break;
181 break;
182 } while(true);
302 }
183 }
303
184
304 // fetch task
305 if (TryDequeue(out unit)) {
306 InvokeUnit(unit);
307 count ++;
308 continue;
309 }
310 Interlocked.Decrement(ref m_activeThreads);
311
312 Console.WriteLine("{0}: Suspend processed({1})", Thread.CurrentThread.ManagedThreadId,count);
313 // entering suspend state
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)
329 SignalThread();
190 SignalThread();
330 GC.SuppressFinalize(this);
191 GC.SuppressFinalize(this);
331 }
192 }
Show file before | Show file after | Hide comments
@@ -66,10 +66,11 namespace Implab.Parallels {
66 var len = Interlocked.Increment(ref m_queueLength);
66 var len = Interlocked.Increment(ref m_queueLength);
67 m_queue.Enqueue(unit);
67 m_queue.Enqueue(unit);
68
68
69 if (len > m_threshold * m_workers) {
69 if (len > m_threshold * PoolSize) {
70 Interlocked.Increment(ref m_workers);
70 StartWorker();
71 GrowPool();
72 }
71 }
72
73 SignalThread();
73 }
74 }
74
75
75 protected override bool TryDequeue(out Action unit) {
76 protected override bool TryDequeue(out Action unit) {
@@ -80,24 +81,6 namespace Implab.Parallels {
80 return false;
81 return false;
81 }
82 }
82
83
83 protected override bool Suspend() {
84 // This override solves race condition
85 // WORKER CLIENT
86 // ---------------------------------------
87 // TryDeque == false
88 // Enqueue(unit), queueLen++
89 // GrowPool? == NO
90 // ActiveThreads--
91 // Suspend
92 // queueLength > 0
93 // continue
94 Thread.MemoryBarrier();
95 if (m_queueLength > 0)
96 return true;
97 Interlocked.Decrement(ref m_workers);
98 return base.Suspend();
99 }
100
101 protected override void InvokeUnit(Action unit) {
84 protected override void InvokeUnit(Action unit) {
102 unit();
85 unit();
103 }
86 }
General Comments 0
You need to be logged in to leave comments. Login now