##// END OF EJS Templates
implab » ImplabNet
RSS
implemented parallel map and foreach for arrays...
cin -
r15:0f982f9b7d4d promises
parent child
Show More
Show file before | Show file after | Hide comments
@@ -0,0 +1,152
1 using System;
2 using System.Collections.Generic;
3 using System.Diagnostics;
4 using System.Linq;
5 using System.Text;
6 using System.Threading;
7
8 namespace Implab.Parallels {
9 public static class ArrayTraits {
10 class ArrayIterator<TSrc> : DispatchPool<int> {
11 readonly Action<TSrc> m_action;
12 readonly TSrc[] m_source;
13 readonly Promise<int> m_promise = new Promise<int>();
14
15 int m_pending;
16 int m_next;
17
18 public ArrayIterator(TSrc[] source, Action<TSrc> action, int threads)
19 : base(threads) {
20
21 Debug.Assert(source != null);
22 Debug.Assert(action != null);
23
24 m_next = 0;
25 m_source = source;
26 m_pending = source.Length;
27 m_action = action;
28
29 m_promise.Anyway(() => Dispose());
30 m_promise.Cancelled(() => Dispose());
31
32 InitPool();
33 }
34
35 public Promise<int> Promise {
36 get {
37 return m_promise;
38 }
39 }
40
41 protected override bool TryDequeue(out int unit) {
42 int index;
43 unit = -1;
44 do {
45 index = m_next;
46 if (index >= m_source.Length)
47 return false;
48 } while (index != Interlocked.CompareExchange(ref m_next, index + 1, index));
49
50 unit = index;
51 return true;
52 }
53
54 protected override void InvokeUnit(int unit) {
55 try {
56 m_action(m_source[unit]);
57 int pending;
58 do {
59 pending = m_pending;
60 } while (pending != Interlocked.CompareExchange(ref m_pending, pending - 1, pending));
61 pending--;
62 if (pending == 0)
63 m_promise.Resolve(m_source.Length);
64 } catch (Exception e) {
65 m_promise.Reject(e);
66 }
67 }
68 }
69
70 class ArrayMapper<TSrc, TDst>: DispatchPool<int> {
71 readonly Func<TSrc, TDst> m_transform;
72 readonly TSrc[] m_source;
73 readonly TDst[] m_dest;
74 readonly Promise<TDst[]> m_promise = new Promise<TDst[]>();
75
76 int m_pending;
77 int m_next;
78
79 public ArrayMapper(TSrc[] source, Func<TSrc, TDst> transform, int threads)
80 : base(threads) {
81
82 Debug.Assert (source != null);
83 Debug.Assert( transform != null);
84
85 m_next = 0;
86 m_source = source;
87 m_dest = new TDst[source.Length];
88 m_pending = source.Length;
89 m_transform = transform;
90
91 m_promise.Anyway(() => Dispose());
92 m_promise.Cancelled(() => Dispose());
93
94 InitPool();
95 }
96
97 public Promise<TDst[]> Promise {
98 get {
99 return m_promise;
100 }
101 }
102
103 protected override bool TryDequeue(out int unit) {
104 int index;
105 unit = -1;
106 do {
107 index = m_next;
108 if (index >= m_source.Length)
109 return false;
110 } while (index != Interlocked.CompareExchange(ref m_next, index + 1, index));
111
112 unit = index;
113 return true;
114 }
115
116 protected override void InvokeUnit(int unit) {
117 try {
118 m_dest[unit] = m_transform(m_source[unit]);
119 int pending;
120 do {
121 pending = m_pending;
122 } while ( pending != Interlocked.CompareExchange(ref m_pending,pending -1, pending));
123 pending --;
124 if (pending == 0)
125 m_promise.Resolve(m_dest);
126 } catch (Exception e) {
127 m_promise.Reject(e);
128 }
129 }
130 }
131
132 public static Promise<TDst[]> ParallelMap<TSrc, TDst> (this TSrc[] source, Func<TSrc,TDst> transform, int threads) {
133 if (source == null)
134 throw new ArgumentNullException("source");
135 if (transform == null)
136 throw new ArgumentNullException("transform");
137
138 var mapper = new ArrayMapper<TSrc, TDst>(source, transform, threads);
139 return mapper.Promise;
140 }
141
142 public static Promise<int> ParallelForEach<TSrc>(this TSrc[] source, Action<TSrc> action, int threads) {
143 if (source == null)
144 throw new ArgumentNullException("source");
145 if (action == null)
146 throw new ArgumentNullException("action");
147
148 var iter = new ArrayIterator<TSrc>(source, action, threads);
149 return iter.Promise;
150 }
151 }
152 }
Show file before | Show file after | Hide comments
@@ -0,0 +1,171
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;
12 int m_runningThreads = 0;
13 int m_maxRunningThreads = 0;
14 int m_suspended = 0;
15 int m_exitRequired = 0;
16 AutoResetEvent m_hasTasks = new AutoResetEvent(false);
17
18 protected DispatchPool(int min, int max) {
19 if (min < 0)
20 throw new ArgumentOutOfRangeException("min");
21 if (max <= 0)
22 throw new ArgumentOutOfRangeException("max");
23
24 if (min > max)
25 min = max;
26 m_minThreads = min;
27 m_maxThreads = max;
28 }
29
30 protected DispatchPool(int threads)
31 : this(threads, threads) {
32 }
33
34 protected DispatchPool() {
35 int maxThreads, maxCP;
36 ThreadPool.GetMaxThreads(out maxThreads, out maxCP);
37
38 m_minThreads = 0;
39 m_maxThreads = maxThreads;
40 }
41
42 protected void InitPool() {
43 for (int i = 0; i < m_minThreads; i++)
44 StartWorker();
45 }
46
47 public int ThreadCount {
48 get {
49 return m_runningThreads;
50 }
51 }
52
53 public int MaxRunningThreads {
54 get {
55 return m_maxRunningThreads;
56 }
57 }
58
59 protected bool IsDisposed {
60 get {
61 return m_exitRequired != 0;
62 }
63 }
64
65 bool StartWorker() {
66 var current = m_runningThreads;
67 // use spins to allocate slot for the new thread
68 do {
69 if (current >= m_maxThreads || m_exitRequired != 0)
70 // no more slots left or the pool has been disposed
71 return false;
72 } while (current != Interlocked.CompareExchange(ref m_runningThreads, current + 1, current));
73
74 m_maxRunningThreads = Math.Max(m_maxRunningThreads, current + 1);
75
76 // slot successfully allocated
77
78 var worker = new Thread(this.Worker);
79 worker.IsBackground = true;
80 worker.Start();
81
82 return true;
83 }
84
85 protected abstract bool TryDequeue(out TUnit unit);
86
87 protected virtual void WakeNewWorker() {
88 if (m_suspended > 0)
89 m_hasTasks.Set();
90 else
91 StartWorker();
92 }
93
94 bool FetchTask(out TUnit unit) {
95 do {
96 // exit if requested
97 if (m_exitRequired != 0) {
98 // release the thread slot
99 int running;
100 do {
101 running = m_runningThreads;
102 } while (running != Interlocked.CompareExchange(ref m_runningThreads, running - 1, running));
103 running--;
104
105 if (running == 0) // it was the last worker
106 m_hasTasks.Dispose();
107 else
108 m_hasTasks.Set(); // release next worker
109 unit = default(TUnit);
110 return false;
111 }
112
113 // fetch task
114 if (TryDequeue(out unit)) {
115 WakeNewWorker();
116 return true;
117 }
118
119 //no tasks left, exit if the thread is no longer needed
120 int runningThreads;
121 bool exit = true;
122 do {
123 runningThreads = m_runningThreads;
124 if (runningThreads <= m_minThreads) {
125 exit = false;
126 break;
127 }
128 } while (runningThreads != Interlocked.CompareExchange(ref m_runningThreads, runningThreads - 1, runningThreads));
129
130 if (exit) {
131 Interlocked.Decrement(ref m_runningThreads);
132 return false;
133 }
134
135 // keep this thread and wait
136 Interlocked.Increment(ref m_suspended);
137 m_hasTasks.WaitOne();
138 Interlocked.Decrement(ref m_suspended);
139 } while (true);
140 }
141
142 protected abstract void InvokeUnit(TUnit unit);
143
144 void Worker() {
145 TUnit unit;
146 while (FetchTask(out unit))
147 InvokeUnit(unit);
148 }
149
150 protected virtual void Dispose(bool disposing) {
151 if (disposing) {
152 if (m_exitRequired == 0) {
153 if (Interlocked.CompareExchange(ref m_exitRequired, 1, 0) != 0)
154 return;
155
156 // wake sleeping threads
157 m_hasTasks.Set();
158 GC.SuppressFinalize(this);
159 }
160 }
161 }
162
163 public void Dispose() {
164 Dispose(true);
165 }
166
167 ~DispatchPool() {
168 Dispose(false);
169 }
170 }
171 }
Show file before | Show file after | Hide comments
@@ -4,71 +4,64 using System.Reflection;
4 using System.Threading;
4 using System.Threading;
5 using Implab.Parallels;
5 using Implab.Parallels;
6
6
7 namespace Implab.Test
7 namespace Implab.Test {
8 {
8 [TestClass]
9 [TestClass]
9 public class AsyncTests {
10 public class AsyncTests
10 [TestMethod]
11 {
11 public void ResolveTest() {
12 [TestMethod]
12 int res = -1;
13 public void ResolveTest ()
13 var p = new Promise<int>();
14 {
14 p.Then(x => res = x);
15 int res = -1;
15 p.Resolve(100);
16 var p = new Promise<int> ();
17 p.Then (x => res = x);
18 p.Resolve (100);
19
16
20 Assert.AreEqual (res, 100);
17 Assert.AreEqual(res, 100);
21 }
18 }
22
19
23 [TestMethod]
20 [TestMethod]
24 public void RejectTest ()
21 public void RejectTest() {
25 {
22 int res = -1;
26 int res = -1;
23 Exception err = null;
27 Exception err = null;
28
24
29 var p = new Promise<int> ();
25 var p = new Promise<int>();
30 p.Then (x => res = x, e => err = e);
26 p.Then(x => res = x, e => err = e);
31 p.Reject (new ApplicationException ("error"));
27 p.Reject(new ApplicationException("error"));
32
28
33 Assert.AreEqual (res, -1);
29 Assert.AreEqual(res, -1);
34 Assert.AreEqual (err.Message, "error");
30 Assert.AreEqual(err.Message, "error");
35
31
36 }
32 }
37
33
38 [TestMethod]
34 [TestMethod]
39 public void JoinSuccessTest ()
35 public void JoinSuccessTest() {
40 {
36 var p = new Promise<int>();
41 var p = new Promise<int> ();
37 p.Resolve(100);
42 p.Resolve (100);
38 Assert.AreEqual(p.Join(), 100);
43 Assert.AreEqual (p.Join (), 100);
39 }
44 }
45
40
46 [TestMethod]
41 [TestMethod]
47 public void JoinFailTest ()
42 public void JoinFailTest() {
48 {
43 var p = new Promise<int>();
49 var p = new Promise<int> ();
44 p.Reject(new ApplicationException("failed"));
50 p.Reject (new ApplicationException ("failed"));
51
45
52 try {
46 try {
53 p.Join ();
47 p.Join();
54 throw new ApplicationException ("WRONG!");
48 throw new ApplicationException("WRONG!");
55 } catch (TargetInvocationException err) {
49 } catch (TargetInvocationException err) {
56 Assert.AreEqual (err.InnerException.Message, "failed");
50 Assert.AreEqual(err.InnerException.Message, "failed");
57 } catch {
51 } catch {
58 Assert.Fail ("Got wrong excaption");
52 Assert.Fail("Got wrong excaption");
59 }
53 }
60 }
54 }
61
55
62 [TestMethod]
56 [TestMethod]
63 public void MapTest ()
57 public void MapTest() {
64 {
58 var p = new Promise<int>();
65 var p = new Promise<int> ();
66
59
67 var p2 = p.Map (x => x.ToString ());
60 var p2 = p.Map(x => x.ToString());
68 p.Resolve (100);
61 p.Resolve(100);
69
62
70 Assert.AreEqual (p2.Join (), "100");
63 Assert.AreEqual(p2.Join(), "100");
71 }
64 }
72
65
73 [TestMethod]
66 [TestMethod]
74 public void FixErrorTest() {
67 public void FixErrorTest() {
@@ -82,65 +75,90 namespace Implab.Test
82 }
75 }
83
76
84 [TestMethod]
77 [TestMethod]
85 public void ChainTest ()
78 public void ChainTest() {
86 {
79 var p1 = new Promise<int>();
87 var p1 = new Promise<int> ();
88
80
89 var p3 = p1.Chain (x => {
81 var p3 = p1.Chain(x => {
90 var p2 = new Promise<string> ();
82 var p2 = new Promise<string>();
91 p2.Resolve (x.ToString ());
83 p2.Resolve(x.ToString());
92 return p2;
84 return p2;
93 });
85 });
94
86
95 p1.Resolve (100);
87 p1.Resolve(100);
96
88
97 Assert.AreEqual (p3.Join (), "100");
89 Assert.AreEqual(p3.Join(), "100");
98 }
90 }
99
91
100 [TestMethod]
92 [TestMethod]
101 public void PoolTest ()
93 public void PoolTest() {
102 {
94 var pid = Thread.CurrentThread.ManagedThreadId;
103 var pid = Thread.CurrentThread.ManagedThreadId;
95 var p = AsyncPool.Invoke(() => Thread.CurrentThread.ManagedThreadId);
104 var p = AsyncPool.Invoke (() => Thread.CurrentThread.ManagedThreadId);
105
96
106 Assert.AreNotEqual (pid, p.Join ());
97 Assert.AreNotEqual(pid, p.Join());
107 }
98 }
108
99
109 [TestMethod]
100 [TestMethod]
110 public void WorkerPoolSizeTest() {
101 public void WorkerPoolSizeTest() {
111 var pool = new WorkerPool(5,10);
102 var pool = new WorkerPool(5, 10);
112
103
113 Assert.AreEqual(5, pool.ThreadCount);
104 Assert.AreEqual(5, pool.ThreadCount);
114
105
115 pool.Invoke(() => { Thread.Sleep(1000); return 10; });
106 pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
116 pool.Invoke(() => { Thread.Sleep(1000); return 10; });
107 pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
117 pool.Invoke(() => { Thread.Sleep(1000); return 10; });
108 pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
118
109
119 Assert.AreEqual(5, pool.ThreadCount);
110 Assert.AreEqual(5, pool.ThreadCount);
120
111
121 for (int i = 0; i < 100; i++)
112 for (int i = 0; i < 100; i++)
122 pool.Invoke(() => { Thread.Sleep(1000); return 10; });
113 pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
114 Thread.Sleep(100);
123 Assert.AreEqual(10, pool.ThreadCount);
115 Assert.AreEqual(10, pool.ThreadCount);
116
117 pool.Dispose();
124 }
118 }
125
119
126 [TestMethod]
120 [TestMethod]
127 public void WorkerPoolCorrectTest() {
121 public void WorkerPoolCorrectTest() {
128 var pool = new WorkerPool(5, 20);
122 var pool = new WorkerPool();
123
124 int iterations = 1000;
125 int pending = iterations;
126 var stop = new ManualResetEvent(false);
129
127
130 var count = 0;
128 var count = 0;
131 for (int i = 0; i < 1000; i++)
129 for (int i = 0; i < iterations; i++) {
132 pool
130 pool
133 .Invoke(() => 1)
131 .Invoke(() => 1)
134 .Then(x => Interlocked.Add(ref count, x));
132 .Then(x => Interlocked.Add(ref count, x))
133 .Then(x => Math.Log10(x))
134 .Anyway(() => {
135 Interlocked.Decrement(ref pending);
136 if (pending == 0)
137 stop.Set();
138 });
139 }
140
141 stop.WaitOne();
135
142
136 Assert.AreEqual(1000, count);
143 Assert.AreEqual(iterations, count);
144 Console.WriteLine("Max threads: {0}", pool.MaxRunningThreads);
145 pool.Dispose();
146
147 }
148
149 [TestMethod]
150 public void WorkerPoolDisposeTest() {
151 var pool = new WorkerPool(5, 20);
152 Assert.AreEqual(5, pool.ThreadCount);
153 pool.Dispose();
154 Thread.Sleep(100);
155 Assert.AreEqual(0, pool.ThreadCount);
156 pool.Dispose();
137 }
157 }
138
158
139 [TestMethod]
159 [TestMethod]
140 public void MTQueueTest() {
160 public void MTQueueTest() {
141 var queue = new MTQueue<int>();
161 var queue = new MTQueue<int>();
142 var pool = new WorkerPool(5, 20);
143
144 int res;
162 int res;
145
163
146 queue.Enqueue(10);
164 queue.Enqueue(10);
@@ -169,33 +187,27 namespace Implab.Test
169 var wn = i;
187 var wn = i;
170 AsyncPool
188 AsyncPool
171 .InvokeNewThread(() => {
189 .InvokeNewThread(() => {
172 Console.WriteLine("Started writer: {0}", wn);
173 for (int ii = 0; ii < itemsPerWriter; ii++) {
190 for (int ii = 0; ii < itemsPerWriter; ii++) {
174 queue.Enqueue(1);
191 queue.Enqueue(1);
175 Thread.Sleep(1);
176 }
192 }
177 Console.WriteLine("Stopped writer: {0}", wn);
178 return 1;
193 return 1;
179 })
194 })
180 .Then(x => Interlocked.Decrement(ref writers) );
195 .Anyway(() => Interlocked.Decrement(ref writers));
181 }
196 }
182
197
183 for (int i = 0; i < 10; i++) {
198 for (int i = 0; i < 10; i++) {
184 Interlocked.Increment(ref readers);
199 Interlocked.Increment(ref readers);
185 var wn = i;
200 var wn = i;
186 AsyncPool
201 AsyncPool
187 .InvokeNewThread(() => {
202 .InvokeNewThread(() => {
188 int t;
203 int t;
189 Console.WriteLine("Started reader: {0}", wn);
190 do {
204 do {
191 while (queue.TryDequeue(out t))
205 while (queue.TryDequeue(out t))
192 Interlocked.Add(ref total, t);
206 Interlocked.Add(ref total, t);
193 Thread.Sleep(0);
194 } while (writers > 0);
207 } while (writers > 0);
195 Console.WriteLine("Stopped reader: {0}", wn);
196 return 1;
208 return 1;
197 })
209 })
198 .Then(x => {
210 .Anyway(() => {
199 Interlocked.Decrement(ref readers);
211 Interlocked.Decrement(ref readers);
200 if (readers == 0)
212 if (readers == 0)
201 stop.Set();
213 stop.Set();
@@ -208,6 +220,55 namespace Implab.Test
208 }
220 }
209
221
210 [TestMethod]
222 [TestMethod]
223 public void ParallelMapTest() {
224
225 int count = 100000;
226
227 double[] args = new double[count];
228 var rand = new Random();
229
230 for (int i = 0; i < count; i++)
231 args[i] = rand.NextDouble();
232
233 var t = Environment.TickCount;
234 var res = args.ParallelMap(x => Math.Sin(x*x), 4).Join();
235
236 Console.WriteLine("Map complete in {0} ms", Environment.TickCount - t);
237
238 t = Environment.TickCount;
239 for (int i = 0; i < count; i++)
240 Assert.AreEqual(Math.Sin(args[i] * args[i]), res[i]);
241 Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
242 }
243
244 [TestMethod]
245 public void ParallelForEachTest() {
246
247 int count = 100000;
248
249 int[] args = new int[count];
250 var rand = new Random();
251
252 for (int i = 0; i < count; i++)
253 args[i] = (int)(rand.NextDouble() * 100);
254
255 int result = 0;
256
257 var t = Environment.TickCount;
258 args.ParallelForEach(x => Interlocked.Add(ref result, x), 4).Join();
259
260 Console.WriteLine("Iteration complete in {0} ms, result: {1}", Environment.TickCount - t, result);
261
262 int result2 = 0;
263
264 t = Environment.TickCount;
265 for (int i = 0; i < count; i++)
266 result2 += args[i];
267 Assert.AreEqual(result2, result);
268 Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
269 }
270
271 [TestMethod]
211 public void ComplexCase1Test() {
272 public void ComplexCase1Test() {
212 var flags = new bool[3];
273 var flags = new bool[3];
213
274
@@ -219,7 +280,7 namespace Implab.Test
219 .Chain(x =>
280 .Chain(x =>
220 PromiseHelper
281 PromiseHelper
221 .Sleep(200, "Hi, " + x)
282 .Sleep(200, "Hi, " + x)
222 .Map( y => y )
283 .Map(y => y)
223 .Cancelled(() => flags[1] = true)
284 .Cancelled(() => flags[1] = true)
224 )
285 )
225 .Cancelled(() => flags[2] = true);
286 .Cancelled(() => flags[2] = true);
@@ -228,13 +289,13 namespace Implab.Test
228 try {
289 try {
229 Assert.AreEqual(p.Join(), "Hi, Alan");
290 Assert.AreEqual(p.Join(), "Hi, Alan");
230 Assert.Fail("Shouldn't get here");
291 Assert.Fail("Shouldn't get here");
231 } catch(OperationCanceledException) {
292 } catch (OperationCanceledException) {
232 }
293 }
233
294
234 Assert.IsFalse(flags[0]);
295 Assert.IsFalse(flags[0]);
235 Assert.IsTrue(flags[1]);
296 Assert.IsTrue(flags[1]);
236 Assert.IsTrue(flags[2]);
297 Assert.IsTrue(flags[2]);
237 }
298 }
238 }
299 }
239 }
300 }
240
301
Show file before | Show file after | Hide comments
1 NO CONTENT: modified file, binary diff hidden
NO CONTENT: modified file, binary diff hidden
Show file before | Show file after | Hide comments
@@ -38,6 +38,8
38 <Compile Include="IPromise.cs" />
38 <Compile Include="IPromise.cs" />
39 <Compile Include="ITaskController.cs" />
39 <Compile Include="ITaskController.cs" />
40 <Compile Include="ManagedPromise.cs" />
40 <Compile Include="ManagedPromise.cs" />
41 <Compile Include="Parallels\DispatchPool.cs" />
42 <Compile Include="Parallels\ArrayTraits.cs" />
41 <Compile Include="Parallels\MTQueue.cs" />
43 <Compile Include="Parallels\MTQueue.cs" />
42 <Compile Include="Parallels\WorkerPool.cs" />
44 <Compile Include="Parallels\WorkerPool.cs" />
43 <Compile Include="PromiseState.cs" />
45 <Compile Include="PromiseState.cs" />
Show file before | Show file after | Hide comments
@@ -6,66 +6,35 using System.Threading;
6 using System.Diagnostics;
6 using System.Diagnostics;
7
7
8 namespace Implab.Parallels {
8 namespace Implab.Parallels {
9 public class WorkerPool : IDisposable {
9 public class WorkerPool : DispatchPool<Action> {
10 readonly int m_minThreads;
11 readonly int m_maxThreads;
12 int m_runningThreads;
13 object m_lock = new object();
14
15 bool m_disposed = false;
16
17 // this event will signal that workers can try to fetch a task from queue or the pool has been disposed
18 ManualResetEvent m_hasTasks = new ManualResetEvent(false);
19 Queue<Action> m_queue = new Queue<Action>();
20
10
21 public WorkerPool(int min, int max) {
11 MTQueue<Action> m_queue = new MTQueue<Action>();
22 if (min < 0)
12 int m_queueLength = 0;
23 throw new ArgumentOutOfRangeException("min");
24 if (max <= 0)
25 throw new ArgumentOutOfRangeException("max");
26
13
27 if (min > max)
14 public WorkerPool(int minThreads, int maxThreads)
28 min = max;
15 : base(minThreads, maxThreads) {
29 m_minThreads = min;
16 InitPool();
30 m_maxThreads = max;
31
32 InitPool();
33 }
17 }
34
18
35 public WorkerPool(int max)
19 public WorkerPool(int threads)
36 : this(0, max) {
20 : base(threads) {
21 InitPool();
37 }
22 }
38
23
39 public WorkerPool() {
24 public WorkerPool()
40 int maxThreads, maxCP;
25 : base() {
41 ThreadPool.GetMaxThreads(out maxThreads, out maxCP);
26 InitPool();
42
43 m_minThreads = 0;
44 m_maxThreads = maxThreads;
45
46 InitPool();
47 }
48
49 void InitPool() {
50 for (int i = 0; i < m_minThreads; i++)
51 StartWorker();
52 }
53
54 public int ThreadCount {
55 get {
56 return m_runningThreads;
57 }
58 }
27 }
59
28
60 public Promise<T> Invoke<T>(Func<T> task) {
29 public Promise<T> Invoke<T>(Func<T> task) {
61 if (m_disposed)
62 throw new ObjectDisposedException(ToString());
63 if (task == null)
30 if (task == null)
64 throw new ArgumentNullException("task");
31 throw new ArgumentNullException("task");
32 if (IsDisposed)
33 throw new ObjectDisposedException(ToString());
65
34
66 var promise = new Promise<T>();
35 var promise = new Promise<T>();
67
36
68 var queueLen = EnqueueTask(delegate() {
37 EnqueueTask(delegate() {
69 try {
38 try {
70 promise.Resolve(task());
39 promise.Resolve(task());
71 } catch (Exception e) {
40 } catch (Exception e) {
@@ -73,99 +42,28 namespace Implab.Parallels {
73 }
42 }
74 });
43 });
75
44
76 if (queueLen > 1)
77 StartWorker();
78
79 return promise;
45 return promise;
80 }
46 }
81
47
82 bool StartWorker() {
48 protected void EnqueueTask(Action unit) {
83 var current = m_runningThreads;
49 Debug.Assert(unit != null);
84 // use spins to allocate slot for the new thread
50 Interlocked.Increment(ref m_queueLength);
85 do {
51 m_queue.Enqueue(unit);
86 if (current >= m_maxThreads)
52 // if there are sleeping threads in the pool wake one
87 // no more slots left
53 // probably this will lead a dry run
88 return false;
54 WakeNewWorker();
89 } while (current != Interlocked.CompareExchange(ref m_runningThreads, current + 1, current));
90
91 // slot successfully allocated
92
93 var worker = new Thread(this.Worker);
94 worker.IsBackground = true;
95 worker.Start();
96
97 return true;
98 }
99
100 int EnqueueTask(Action task) {
101 Debug.Assert(task != null);
102 lock (m_queue) {
103 m_queue.Enqueue(task);
104 m_hasTasks.Set();
105 return m_queue.Count;
106 }
107 }
55 }
108
56
109 bool FetchTask(out Action task) {
57 protected override bool TryDequeue(out Action unit) {
110 task = null;
58 if (m_queue.TryDequeue(out unit)) {
111
59 Interlocked.Decrement(ref m_queueLength);
112 while (true) {
60 return true;
113
114 m_hasTasks.WaitOne();
115
116 if (m_disposed)
117 return false;
118
119 lock (m_queue) {
120 if (m_queue.Count > 0) {
121 task = m_queue.Dequeue();
122 return true;
123 }
124
125 // no tasks left
126 // signal that no more tasks left, current lock ensures that this event won't suppress newly added task
127 m_hasTasks.Reset();
128 }
129
130 bool exit = true;
131
132 var current = m_runningThreads;
133 do {
134 if (current <= m_minThreads) {
135 exit = false; // this thread should return and wait for the new events
136 break;
137 }
138 } while (current != Interlocked.CompareExchange(ref m_runningThreads, current - 1, current));
139
140 if (exit)
141 return false;
142 }
61 }
62 return false;
143 }
63 }
144
64
145 void Worker() {
65 protected override void InvokeUnit(Action unit) {
146 Action task;
66 unit();
147 while (FetchTask(out task))
148 task();
149 }
150
151 protected virtual void Dispose(bool disposing) {
152 if (disposing) {
153 lock (m_lock) {
154 if (m_disposed)
155 return;
156 m_disposed = true;
157 }
158 m_hasTasks.Set();
159 GC.SuppressFinalize(this);
160 }
161 }
162
163 public void Dispose() {
164 Dispose(true);
165 }
166
167 ~WorkerPool() {
168 Dispose(false);
169 }
67 }
170 }
68 }
171 }
69 }
Show file before | Show file after | Hide comments
@@ -539,5 +539,6 namespace Implab {
539
539
540 return result;
540 return result;
541 }
541 }
542
542 }
543 }
543 }
544 }
General Comments 0
You need to be logged in to leave comments. Login now