##// 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,14 +4,11 using System.Reflection;
4 4 using System.Threading;
5 5 using Implab.Parallels;
6 6
7 namespace Implab.Test
8 {
7 namespace Implab.Test {
9 8 [TestClass]
10 public class AsyncTests
11 {
9 public class AsyncTests {
12 10 [TestMethod]
13 public void ResolveTest ()
14 {
11 public void ResolveTest() {
15 12 int res = -1;
16 13 var p = new Promise<int> ();
17 14 p.Then (x => res = x);
@@ -21,8 +18,7 namespace Implab.Test
21 18 }
22 19
23 20 [TestMethod]
24 public void RejectTest ()
25 {
21 public void RejectTest() {
26 22 int res = -1;
27 23 Exception err = null;
28 24
@@ -36,16 +32,14 namespace Implab.Test
36 32 }
37 33
38 34 [TestMethod]
39 public void JoinSuccessTest ()
40 {
35 public void JoinSuccessTest() {
41 36 var p = new Promise<int> ();
42 37 p.Resolve (100);
43 38 Assert.AreEqual (p.Join (), 100);
44 39 }
45 40
46 41 [TestMethod]
47 public void JoinFailTest ()
48 {
42 public void JoinFailTest() {
49 43 var p = new Promise<int> ();
50 44 p.Reject (new ApplicationException ("failed"));
51 45
@@ -60,8 +54,7 namespace Implab.Test
60 54 }
61 55
62 56 [TestMethod]
63 public void MapTest ()
64 {
57 public void MapTest() {
65 58 var p = new Promise<int> ();
66 59
67 60 var p2 = p.Map (x => x.ToString ());
@@ -82,8 +75,7 namespace Implab.Test
82 75 }
83 76
84 77 [TestMethod]
85 public void ChainTest ()
86 {
78 public void ChainTest() {
87 79 var p1 = new Promise<int> ();
88 80
89 81 var p3 = p1.Chain (x => {
@@ -98,8 +90,7 namespace Implab.Test
98 90 }
99 91
100 92 [TestMethod]
101 public void PoolTest ()
102 {
93 public void PoolTest() {
103 94 var pid = Thread.CurrentThread.ManagedThreadId;
104 95 var p = AsyncPool.Invoke (() => Thread.CurrentThread.ManagedThreadId);
105 96
@@ -112,35 +103,62 namespace Implab.Test
112 103
113 104 Assert.AreEqual(5, pool.ThreadCount);
114 105
115 pool.Invoke(() => { Thread.Sleep(1000); return 10; });
116 pool.Invoke(() => { Thread.Sleep(1000); return 10; });
117 pool.Invoke(() => { Thread.Sleep(1000); return 10; });
106 pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
107 pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
108 pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
118 109
119 110 Assert.AreEqual(5, pool.ThreadCount);
120 111
121 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 115 Assert.AreEqual(10, pool.ThreadCount);
116
117 pool.Dispose();
124 118 }
125 119
126 120 [TestMethod]
127 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 128 var count = 0;
131 for (int i = 0; i < 1000; i++)
129 for (int i = 0; i < iterations; i++) {
132 130 pool
133 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 159 [TestMethod]
140 160 public void MTQueueTest() {
141 161 var queue = new MTQueue<int>();
142 var pool = new WorkerPool(5, 20);
143
144 162 int res;
145 163
146 164 queue.Enqueue(10);
@@ -169,15 +187,12 namespace Implab.Test
169 187 var wn = i;
170 188 AsyncPool
171 189 .InvokeNewThread(() => {
172 Console.WriteLine("Started writer: {0}", wn);
173 190 for (int ii = 0; ii < itemsPerWriter; ii++) {
174 191 queue.Enqueue(1);
175 Thread.Sleep(1);
176 192 }
177 Console.WriteLine("Stopped writer: {0}", wn);
178 193 return 1;
179 194 })
180 .Then(x => Interlocked.Decrement(ref writers) );
195 .Anyway(() => Interlocked.Decrement(ref writers));
181 196 }
182 197
183 198 for (int i = 0; i < 10; i++) {
@@ -186,16 +201,13 namespace Implab.Test
186 201 AsyncPool
187 202 .InvokeNewThread(() => {
188 203 int t;
189 Console.WriteLine("Started reader: {0}", wn);
190 204 do {
191 205 while (queue.TryDequeue(out t))
192 206 Interlocked.Add(ref total, t);
193 Thread.Sleep(0);
194 207 } while (writers > 0);
195 Console.WriteLine("Stopped reader: {0}", wn);
196 208 return 1;
197 209 })
198 .Then(x => {
210 .Anyway(() => {
199 211 Interlocked.Decrement(ref readers);
200 212 if (readers == 0)
201 213 stop.Set();
@@ -208,6 +220,55 namespace Implab.Test
208 220 }
209 221
210 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 272 public void ComplexCase1Test() {
212 273 var flags = new bool[3];
213 274
Show file before | Show file after | Hide comments
1 NO CONTENT: modified file, binary diff hidden
Show file before | Show file after | Hide comments
@@ -38,6 +38,8
38 38 <Compile Include="IPromise.cs" />
39 39 <Compile Include="ITaskController.cs" />
40 40 <Compile Include="ManagedPromise.cs" />
41 <Compile Include="Parallels\DispatchPool.cs" />
42 <Compile Include="Parallels\ArrayTraits.cs" />
41 43 <Compile Include="Parallels\MTQueue.cs" />
42 44 <Compile Include="Parallels\WorkerPool.cs" />
43 45 <Compile Include="PromiseState.cs" />
Show file before | Show file after | Hide comments
@@ -6,66 +6,35 using System.Threading;
6 6 using System.Diagnostics;
7 7
8 8 namespace Implab.Parallels {
9 public class WorkerPool : IDisposable {
10 readonly int m_minThreads;
11 readonly int m_maxThreads;
12 int m_runningThreads;
13 object m_lock = new object();
9 public class WorkerPool : DispatchPool<Action> {
14 10
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>();
11 MTQueue<Action> m_queue = new MTQueue<Action>();
12 int m_queueLength = 0;
20 13
21 public WorkerPool(int min, int max) {
22 if (min < 0)
23 throw new ArgumentOutOfRangeException("min");
24 if (max <= 0)
25 throw new ArgumentOutOfRangeException("max");
14 public WorkerPool(int minThreads, int maxThreads)
15 : base(minThreads, maxThreads) {
16 InitPool();
17 }
26 18
27 if (min > max)
28 min = max;
29 m_minThreads = min;
30 m_maxThreads = max;
31
19 public WorkerPool(int threads)
20 : base(threads) {
32 21 InitPool();
33 22 }
34 23
35 public WorkerPool(int max)
36 : this(0, max) {
37 }
38
39 public WorkerPool() {
40 int maxThreads, maxCP;
41 ThreadPool.GetMaxThreads(out maxThreads, out maxCP);
42
43 m_minThreads = 0;
44 m_maxThreads = maxThreads;
45
24 public WorkerPool()
25 : base() {
46 26 InitPool();
47 27 }
48 28
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 }
59
60 29 public Promise<T> Invoke<T>(Func<T> task) {
61 if (m_disposed)
62 throw new ObjectDisposedException(ToString());
63 30 if (task == null)
64 31 throw new ArgumentNullException("task");
32 if (IsDisposed)
33 throw new ObjectDisposedException(ToString());
65 34
66 35 var promise = new Promise<T>();
67 36
68 var queueLen = EnqueueTask(delegate() {
37 EnqueueTask(delegate() {
69 38 try {
70 39 promise.Resolve(task());
71 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 45 return promise;
80 46 }
81 47
82 bool StartWorker() {
83 var current = m_runningThreads;
84 // use spins to allocate slot for the new thread
85 do {
86 if (current >= m_maxThreads)
87 // no more slots left
88 return false;
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 }
108
109 bool FetchTask(out Action task) {
110 task = null;
111
112 while (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;
48 protected void EnqueueTask(Action unit) {
49 Debug.Assert(unit != null);
50 Interlocked.Increment(ref m_queueLength);
51 m_queue.Enqueue(unit);
52 // if there are sleeping threads in the pool wake one
53 // probably this will lead a dry run
54 WakeNewWorker();
123 55 }
124 56
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();
57 protected override bool TryDequeue(out Action unit) {
58 if (m_queue.TryDequeue(out unit)) {
59 Interlocked.Decrement(ref m_queueLength);
60 return true;
128 61 }
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 62 return false;
142 63 }
143 }
144 64
145 void Worker() {
146 Action task;
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);
65 protected override void InvokeUnit(Action unit) {
66 unit();
169 67 }
170 68 }
171 69 }
Show file before | Show file after | Hide comments
@@ -539,5 +539,6 namespace Implab {
539 539
540 540 return result;
541 541 }
542
542 543 }
543 544 }
General Comments 0
You need to be logged in to leave comments. Login now