annotate 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
Ignore whitespace changes - Everywhere: Within whitespace: At end of lines:
rev   line source
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
1 using System;
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
2 using System.Collections.Generic;
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
3 using System.Linq;
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
4 using System.Text;
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
5 using System.Threading;
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
6 using System.Diagnostics;
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
7
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
8 namespace Implab.Parallels {
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
9 public abstract class DispatchPool<TUnit> : IDisposable {
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
10 readonly int m_minThreadsLimit;
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
11 readonly int m_maxThreadsLimit;
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
12 readonly int m_releaseTimeout = 1000; // the timeout while the working thread will wait for the new tasks before exit
21
6a56df4ec59e DispatchPool works again, but performance is poor in some cases
cin
parents: 20
diff changeset
13
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
14 int m_threads = 0; // the current size of the pool
21
6a56df4ec59e DispatchPool works again, but performance is poor in some cases
cin
parents: 20
diff changeset
15 int m_maxRunningThreads = 0; // the meximum reached size of the pool
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
16 int m_exit = 0; // the pool is going to shutdown, all unused workers are released
80
4f20870d0816 added memory barriers
cin
parents: 41
diff changeset
17
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
18 readonly object m_signal = new object(); // used to pulse waiting threads
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
19
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
20 protected DispatchPool(int min, int max) {
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
21 if (min < 0)
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
22 throw new ArgumentOutOfRangeException("min");
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
23 if (max <= 0)
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
24 throw new ArgumentOutOfRangeException("max");
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
25
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
26 if (min > max)
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
27 min = max;
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
28 m_minThreadsLimit = min;
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
29 m_maxThreadsLimit = max;
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
30 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
31
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
32 protected DispatchPool(int threads)
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
33 : this(threads, threads) {
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
34 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
35
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
36 protected DispatchPool() {
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
37 int maxThreads, maxCP;
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
38 ThreadPool.GetMaxThreads(out maxThreads, out maxCP);
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
39
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
40 m_minThreadsLimit = 0;
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
41 m_maxThreadsLimit = maxThreads;
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
42 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
43
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
44 protected void InitPool() {
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
45 for (int i = 0; i < m_minThreadsLimit; i++)
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
46 StartWorker();
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
47 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
48
20
1c3b3d518480 refactoring, sync
cin
parents: 17
diff changeset
49 public int PoolSize {
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
50 get {
80
4f20870d0816 added memory barriers
cin
parents: 41
diff changeset
51 Thread.MemoryBarrier();
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
52 return m_threads;
20
1c3b3d518480 refactoring, sync
cin
parents: 17
diff changeset
53 }
1c3b3d518480 refactoring, sync
cin
parents: 17
diff changeset
54 }
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
55
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
56 public int MaxRunningThreads {
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
57 get {
80
4f20870d0816 added memory barriers
cin
parents: 41
diff changeset
58 Thread.MemoryBarrier();
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
59 return m_maxRunningThreads;
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
60 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
61 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
62
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
63 protected bool IsDisposed {
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
64 get {
80
4f20870d0816 added memory barriers
cin
parents: 41
diff changeset
65 Thread.MemoryBarrier();
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
66 return m_exit == 1;
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
67 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
68 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
69
17
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
70 protected abstract bool TryDequeue(out TUnit unit);
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
71
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
72 private bool Dequeue(out TUnit unit, int timeout) {
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
73 int ts = Environment.TickCount;
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
74 if (TryDequeue(out unit))
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
75 return true;
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
76 lock (m_signal) {
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
77 while (!TryDequeue(out unit) && m_exit == 0)
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
78 if(!Monitor.Wait(m_signal, Math.Max(0, ts + timeout - Environment.TickCount))) {
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
79 // timeout
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
80 return false;
80
4f20870d0816 added memory barriers
cin
parents: 41
diff changeset
81 }
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
82 // queue item or terminate
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
83 Monitor.Pulse(m_signal);
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
84 if (m_exit == 1)
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
85 return false;
80
4f20870d0816 added memory barriers
cin
parents: 41
diff changeset
86 }
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
87 return true;
22
5a35900264f5 implemented nonblocking wake singnals processing
cin
parents: 21
diff changeset
88 }
5a35900264f5 implemented nonblocking wake singnals processing
cin
parents: 21
diff changeset
89
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
90 protected void SignalThread() {
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
91 lock (m_signal) {
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
92 Monitor.Pulse(m_signal);
30
2fad2d1f4b03 small refactoring, cleanup.
cin
parents: 24
diff changeset
93 }
2fad2d1f4b03 small refactoring, cleanup.
cin
parents: 24
diff changeset
94 }
2fad2d1f4b03 small refactoring, cleanup.
cin
parents: 24
diff changeset
95
17
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
96 #region thread slots traits
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
97
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
98 bool AllocateThreadSlot() {
16
cin
parents: 15
diff changeset
99 int current;
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
100 // use spins to allocate slot for the new thread
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
101 do {
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
102 current = m_threads;
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
103 if (current >= m_maxThreadsLimit || m_exit == 1)
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
104 // no more slots left or the pool has been disposed
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
105 return false;
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
106 } while (current != Interlocked.CompareExchange(ref m_threads, current + 1, current));
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
107
17
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
108 UpdateMaxThreads(current + 1);
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
109
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
110 return true;
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
111 }
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
112
17
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
113 bool AllocateThreadSlot(int desired) {
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
114 if (desired - 1 != Interlocked.CompareExchange(ref m_threads, desired, desired - 1))
17
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
115 return false;
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
116
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
117 UpdateMaxThreads(desired);
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
118
17
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
119 return true;
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
120 }
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
121
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
122 bool ReleaseThreadSlot(out bool last) {
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
123 last = false;
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
124 int current;
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
125 // use spins to release slot for the new thread
80
4f20870d0816 added memory barriers
cin
parents: 41
diff changeset
126 Thread.MemoryBarrier();
17
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
127 do {
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
128 current = m_threads;
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
129 if (current <= m_minThreadsLimit && m_exit == 0)
17
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
130 // the thread is reserved
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
131 return false;
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
132 } while (current != Interlocked.CompareExchange(ref m_threads, current - 1, current));
17
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
133
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
134 last = (current == 1);
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
135
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
136 return true;
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
137 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
138
17
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
139 void UpdateMaxThreads(int count) {
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
140 int max;
16
cin
parents: 15
diff changeset
141 do {
17
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
142 max = m_maxRunningThreads;
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
143 if (max >= count)
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
144 break;
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
145 } while(max != Interlocked.CompareExchange(ref m_maxRunningThreads, count, max));
16
cin
parents: 15
diff changeset
146 }
cin
parents: 15
diff changeset
147
17
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
148 #endregion
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
149
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
150 protected bool StartWorker() {
17
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
151 if (AllocateThreadSlot()) {
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
152 // slot successfully allocated
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
153 var worker = new Thread(this.Worker);
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
154 worker.IsBackground = true;
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
155 worker.Start();
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
156
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
157 return true;
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
158 } else {
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
159 return false;
7cd4a843b4e4 Improved worker pool
cin
parents: 16
diff changeset
160 }
16
cin
parents: 15
diff changeset
161 }
cin
parents: 15
diff changeset
162
21
6a56df4ec59e DispatchPool works again, but performance is poor in some cases
cin
parents: 20
diff changeset
163 protected abstract void InvokeUnit(TUnit unit);
6a56df4ec59e DispatchPool works again, but performance is poor in some cases
cin
parents: 20
diff changeset
164
41
2fc0fbe7d58b Added TraceContext support to array traits
cin
parents: 34
diff changeset
165 protected virtual void Worker() {
21
6a56df4ec59e DispatchPool works again, but performance is poor in some cases
cin
parents: 20
diff changeset
166 TUnit unit;
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
167 bool last;
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
168 do {
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
169 while (Dequeue(out unit, m_releaseTimeout)) {
21
6a56df4ec59e DispatchPool works again, but performance is poor in some cases
cin
parents: 20
diff changeset
170 InvokeUnit(unit);
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
171 }
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
172 if(!ReleaseThreadSlot(out last))
21
6a56df4ec59e DispatchPool works again, but performance is poor in some cases
cin
parents: 20
diff changeset
173 continue;
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
174 // queue may be not empty
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
175 if (last && TryDequeue(out unit)) {
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
176 InvokeUnit(unit);
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
177 if (AllocateThreadSlot(1))
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
178 continue;
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
179 // we can safely exit since pool is alive
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
180 }
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
181 break;
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
182 } while(true);
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
183 }
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
184
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
185
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
186 protected virtual void Dispose(bool disposing) {
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
187 if (disposing) {
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
188 if (0 == Interlocked.CompareExchange(ref m_exit, 1, 0)) { // implies memory barrier
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
189 // wake sleeping threads
81
2c5631b43c7d dispatch pool rewritten
cin
parents: 80
diff changeset
190 SignalThread();
15
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
191 GC.SuppressFinalize(this);
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
192 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
193 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
194 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
195
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
196 public void Dispose() {
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
197 Dispose(true);
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
198 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
199
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
200 ~DispatchPool() {
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
201 Dispose(false);
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
202 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
203 }
0f982f9b7d4d implemented parallel map and foreach for arrays
cin
parents:
diff changeset
204 }