##// 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
@@ -1,240 +1,301
1 1 using System;
2 2 using Microsoft.VisualStudio.TestTools.UnitTesting;
3 3 using System.Reflection;
4 4 using System.Threading;
5 5 using Implab.Parallels;
6 6
7 namespace Implab.Test
8 {
9 [TestClass]
10 public class AsyncTests
11 {
12 [TestMethod]
13 public void ResolveTest ()
14 {
15 int res = -1;
16 var p = new Promise<int> ();
17 p.Then (x => res = x);
18 p.Resolve (100);
7 namespace Implab.Test {
8 [TestClass]
9 public class AsyncTests {
10 [TestMethod]
11 public void ResolveTest() {
12 int res = -1;
13 var p = new Promise<int>();
14 p.Then(x => res = x);
15 p.Resolve(100);
19 16
20 Assert.AreEqual (res, 100);
21 }
17 Assert.AreEqual(res, 100);
18 }
22 19
23 20 [TestMethod]
24 public void RejectTest ()
25 {
26 int res = -1;
27 Exception err = null;
21 public void RejectTest() {
22 int res = -1;
23 Exception err = null;
28 24
29 var p = new Promise<int> ();
30 p.Then (x => res = x, e => err = e);
31 p.Reject (new ApplicationException ("error"));
25 var p = new Promise<int>();
26 p.Then(x => res = x, e => err = e);
27 p.Reject(new ApplicationException("error"));
32 28
33 Assert.AreEqual (res, -1);
34 Assert.AreEqual (err.Message, "error");
29 Assert.AreEqual(res, -1);
30 Assert.AreEqual(err.Message, "error");
35 31
36 }
32 }
37 33
38 34 [TestMethod]
39 public void JoinSuccessTest ()
40 {
41 var p = new Promise<int> ();
42 p.Resolve (100);
43 Assert.AreEqual (p.Join (), 100);
44 }
35 public void JoinSuccessTest() {
36 var p = new Promise<int>();
37 p.Resolve(100);
38 Assert.AreEqual(p.Join(), 100);
39 }
45 40
46 41 [TestMethod]
47 public void JoinFailTest ()
48 {
49 var p = new Promise<int> ();
50 p.Reject (new ApplicationException ("failed"));
42 public void JoinFailTest() {
43 var p = new Promise<int>();
44 p.Reject(new ApplicationException("failed"));
51 45
52 try {
53 p.Join ();
54 throw new ApplicationException ("WRONG!");
55 } catch (TargetInvocationException err) {
56 Assert.AreEqual (err.InnerException.Message, "failed");
57 } catch {
58 Assert.Fail ("Got wrong excaption");
59 }
60 }
46 try {
47 p.Join();
48 throw new ApplicationException("WRONG!");
49 } catch (TargetInvocationException err) {
50 Assert.AreEqual(err.InnerException.Message, "failed");
51 } catch {
52 Assert.Fail("Got wrong excaption");
53 }
54 }
61 55
62 56 [TestMethod]
63 public void MapTest ()
64 {
65 var p = new Promise<int> ();
57 public void MapTest() {
58 var p = new Promise<int>();
66 59
67 var p2 = p.Map (x => x.ToString ());
68 p.Resolve (100);
60 var p2 = p.Map(x => x.ToString());
61 p.Resolve(100);
69 62
70 Assert.AreEqual (p2.Join (), "100");
71 }
63 Assert.AreEqual(p2.Join(), "100");
64 }
72 65
73 66 [TestMethod]
74 67 public void FixErrorTest() {
75 68 var p = new Promise<int>();
76 69
77 70 var p2 = p.Error(e => 101);
78 71
79 72 p.Reject(new Exception());
80 73
81 74 Assert.AreEqual(p2.Join(), 101);
82 75 }
83 76
84 77 [TestMethod]
85 public void ChainTest ()
86 {
87 var p1 = new Promise<int> ();
78 public void ChainTest() {
79 var p1 = new Promise<int>();
88 80
89 var p3 = p1.Chain (x => {
90 var p2 = new Promise<string> ();
91 p2.Resolve (x.ToString ());
92 return p2;
93 });
81 var p3 = p1.Chain(x => {
82 var p2 = new Promise<string>();
83 p2.Resolve(x.ToString());
84 return p2;
85 });
94 86
95 p1.Resolve (100);
87 p1.Resolve(100);
96 88
97 Assert.AreEqual (p3.Join (), "100");
98 }
89 Assert.AreEqual(p3.Join(), "100");
90 }
99 91
100 92 [TestMethod]
101 public void PoolTest ()
102 {
103 var pid = Thread.CurrentThread.ManagedThreadId;
104 var p = AsyncPool.Invoke (() => Thread.CurrentThread.ManagedThreadId);
93 public void PoolTest() {
94 var pid = Thread.CurrentThread.ManagedThreadId;
95 var p = AsyncPool.Invoke(() => Thread.CurrentThread.ManagedThreadId);
105 96
106 Assert.AreNotEqual (pid, p.Join ());
107 }
97 Assert.AreNotEqual(pid, p.Join());
98 }
108 99
109 100 [TestMethod]
110 101 public void WorkerPoolSizeTest() {
111 var pool = new WorkerPool(5,10);
102 var pool = new WorkerPool(5, 10);
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);
147 165 Assert.IsTrue(queue.TryDequeue(out res));
148 166 Assert.AreEqual(10, res);
149 167 Assert.IsFalse(queue.TryDequeue(out res));
150 168
151 169 for (int i = 0; i < 1000; i++)
152 170 queue.Enqueue(i);
153 171
154 172 for (int i = 0; i < 1000; i++) {
155 173 queue.TryDequeue(out res);
156 174 Assert.AreEqual(i, res);
157 175 }
158 176
159 177 int writers = 0;
160 178 int readers = 0;
161 179 var stop = new ManualResetEvent(false);
162 180 int total = 0;
163 181
164 182 int itemsPerWriter = 1000;
165 183 int writersCount = 3;
166 184
167 185 for (int i = 0; i < writersCount; i++) {
168 186 Interlocked.Increment(ref writers);
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++) {
184 199 Interlocked.Increment(ref readers);
185 200 var wn = i;
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();
202 214 });
203 215 }
204 216
205 217 stop.WaitOne();
206 218
207 219 Assert.AreEqual(itemsPerWriter * writersCount, total);
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
214 275 // op1 (aync 200ms) => op2 (async 200ms) => op3 (sync map)
215 276
216 277 var p = PromiseHelper
217 278 .Sleep(200, "Alan")
218 279 .Cancelled(() => flags[0] = true)
219 280 .Chain(x =>
220 281 PromiseHelper
221 282 .Sleep(200, "Hi, " + x)
222 .Map( y => y )
283 .Map(y => y)
223 284 .Cancelled(() => flags[1] = true)
224 285 )
225 286 .Cancelled(() => flags[2] = true);
226 287 Thread.Sleep(300);
227 288 p.Cancel();
228 289 try {
229 290 Assert.AreEqual(p.Join(), "Hi, Alan");
230 291 Assert.Fail("Shouldn't get here");
231 } catch(OperationCanceledException) {
292 } catch (OperationCanceledException) {
232 293 }
233 294
234 295 Assert.IsFalse(flags[0]);
235 296 Assert.IsTrue(flags[1]);
236 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
Show file before | Show file after | Hide comments
@@ -1,54 +1,56
1 1 ο»Ώ<?xml version="1.0" encoding="utf-8"?>
2 2 <Project DefaultTargets="Build" ToolsVersion="4.0" xmlns="http://schemas.microsoft.com/developer/msbuild/2003">
3 3 <PropertyGroup>
4 4 <Configuration Condition=" '$(Configuration)' == '' ">Debug</Configuration>
5 5 <Platform Condition=" '$(Platform)' == '' ">AnyCPU</Platform>
6 6 <ProductVersion>10.0.0</ProductVersion>
7 7 <SchemaVersion>2.0</SchemaVersion>
8 8 <ProjectGuid>{F550F1F8-8746-4AD0-9614-855F4C4B7F05}</ProjectGuid>
9 9 <OutputType>Library</OutputType>
10 10 <RootNamespace>Implab</RootNamespace>
11 11 <AssemblyName>Implab</AssemblyName>
12 12 </PropertyGroup>
13 13 <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Debug|AnyCPU' ">
14 14 <DebugSymbols>true</DebugSymbols>
15 15 <DebugType>full</DebugType>
16 16 <Optimize>false</Optimize>
17 17 <OutputPath>bin\Debug</OutputPath>
18 18 <DefineConstants>DEBUG;</DefineConstants>
19 19 <ErrorReport>prompt</ErrorReport>
20 20 <WarningLevel>4</WarningLevel>
21 21 <ConsolePause>false</ConsolePause>
22 22 </PropertyGroup>
23 23 <PropertyGroup Condition=" '$(Configuration)|$(Platform)' == 'Release|AnyCPU' ">
24 24 <DebugType>full</DebugType>
25 25 <Optimize>true</Optimize>
26 26 <OutputPath>bin\Release</OutputPath>
27 27 <ErrorReport>prompt</ErrorReport>
28 28 <WarningLevel>4</WarningLevel>
29 29 <ConsolePause>false</ConsolePause>
30 30 </PropertyGroup>
31 31 <ItemGroup>
32 32 <Reference Include="System" />
33 33 </ItemGroup>
34 34 <ItemGroup>
35 35 <Compile Include="ICancellable.cs" />
36 36 <Compile Include="IProgressHandler.cs" />
37 37 <Compile Include="IProgressNotifier.cs" />
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" />
44 46 <Compile Include="TaskController.cs" />
45 47 <Compile Include="ProgressInitEventArgs.cs" />
46 48 <Compile Include="Properties\AssemblyInfo.cs" />
47 49 <Compile Include="Promise.cs" />
48 50 <Compile Include="Parallels\AsyncPool.cs" />
49 51 <Compile Include="Safe.cs" />
50 52 <Compile Include="ValueEventArgs.cs" />
51 53 </ItemGroup>
52 54 <Import Project="$(MSBuildBinPath)\Microsoft.CSharp.targets" />
53 55 <ItemGroup />
54 56 </Project> No newline at end of file
Show file before | Show file after | Hide comments
@@ -1,171 +1,69
1 1 using System;
2 2 using System.Collections.Generic;
3 3 using System.Linq;
4 4 using System.Text;
5 5 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();
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>();
9 public class WorkerPool : DispatchPool<Action> {
20 10
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");
11 MTQueue<Action> m_queue = new MTQueue<Action>();
12 int m_queueLength = 0;
26 13
27 if (min > max)
28 min = max;
29 m_minThreads = min;
30 m_maxThreads = max;
31
32 InitPool();
14 public WorkerPool(int minThreads, int maxThreads)
15 : base(minThreads, maxThreads) {
16 InitPool();
33 17 }
34 18
35 public WorkerPool(int max)
36 : this(0, max) {
19 public WorkerPool(int threads)
20 : base(threads) {
21 InitPool();
37 22 }
38 23
39 public WorkerPool() {
40 int maxThreads, maxCP;
41 ThreadPool.GetMaxThreads(out maxThreads, out maxCP);
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 }
24 public WorkerPool()
25 : base() {
26 InitPool();
58 27 }
59 28
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) {
72 41 promise.Reject(e);
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 }
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();
107 55 }
108 56
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;
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;
57 protected override bool TryDequeue(out Action unit) {
58 if (m_queue.TryDequeue(out unit)) {
59 Interlocked.Decrement(ref m_queueLength);
60 return true;
142 61 }
62 return false;
143 63 }
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
@@ -1,543 +1,544
1 1 using System;
2 2 using System.Collections.Generic;
3 3 using System.Reflection;
4 4 using System.Diagnostics;
5 5 using System.Threading;
6 6
7 7 namespace Implab {
8 8
9 9 public delegate void ErrorHandler(Exception e);
10 10 public delegate T ErrorHandler<out T>(Exception e);
11 11 public delegate void ResultHandler<in T>(T result);
12 12 public delegate TNew ResultMapper<in TSrc, out TNew>(TSrc result);
13 13 public delegate Promise<TNew> ChainedOperation<in TSrc, TNew>(TSrc result);
14 14
15 15 /// <summary>
16 16 /// Класс для асинхронного получСния Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚ΠΎΠ². Π’Π°ΠΊ Π½Π°Π·Ρ‹Π²Π°Π΅ΠΌΠΎΠ΅ "ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅".
17 17 /// </summary>
18 18 /// <typeparam name="T">Π’ΠΈΠΏ ΠΏΠΎΠ»ΡƒΡ‡Π°Π΅ΠΌΠΎΠ³ΠΎ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Π°</typeparam>
19 19 /// <remarks>
20 20 /// <para>БСрвис ΠΏΡ€ΠΈ ΠΎΠ±Ρ€Π°Ρ‰Π΅Π½ΠΈΠΈ ΠΊ Π΅Π³ΠΎ ΠΌΠ΅Ρ‚ΠΎΠ΄Ρƒ Π΄Π°Π΅Ρ‚ ΠΎΠ±Π΅Ρ‰Π°ΠΈΠ½ΠΈΠ΅ ΠΎ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΈΠΈ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ,
21 21 /// ΠΊΠ»ΠΈΠ΅Π½Ρ‚ ΠΏΠΎΠ»ΡƒΡ‡ΠΈΠ² Ρ‚Π°ΠΊΠΎΠ΅ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ ΠΌΠΎΠΆΠ΅Ρ‚ ΡƒΡΡ‚Π°Π½ΠΎΠ²ΠΈΡ‚ΡŒ ряд ΠΎΠ±Ρ€Π°Ρ‚Π½Ρ‹Ρ… Π²Ρ‹Π·ΠΎΠ²ΠΎ для получСния
22 22 /// событий выполнСния обСщания, Ρ‚ΠΎΠ΅ΡΡ‚ΡŒ Π·Π°Π²Π΅Ρ€ΡˆΠ΅Π½ΠΈΡ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ ΠΈ прСдоставлСнии Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚ΠΎΠ².</para>
23 23 /// <para>
24 24 /// ΠžΠ±Π΅Ρ‰Π΅Π½ΠΈΠ΅ ΠΌΠΎΠΆΠ΅Ρ‚ Π±Ρ‹Ρ‚ΡŒ ΠΊΠ°ΠΊ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΎ, Ρ‚Π°ΠΊ ΠΈ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΎ с ошибкой. Для подписки Π½Π°
25 25 /// Π΄Π°Π½Π½Ρ‹Π΅ события ΠΊΠ»ΠΈΠ΅Π½Ρ‚ Π΄ΠΎΠ»ΠΆΠ΅Π½ ΠΈΡΠΏΠΎΠ»ΡŒΠ·ΠΎΠ²Π°Ρ‚ΡŒ ΠΌΠ΅Ρ‚ΠΎΠ΄Ρ‹ <c>Then</c>.
26 26 /// </para>
27 27 /// <para>
28 28 /// БСрвис, Π² свою ΠΎΡ‡Π΅Ρ€Π΅Π΄ΡŒ, ΠΏΠΎ ΠΎΠΊΠΎΠ½Ρ‡Π°Π½ΠΈΡŽ выполнСния ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ (Π²ΠΎΠ·ΠΌΠΎΠΆΠ½ΠΎ с ошибкой),
29 29 /// ΠΈΡΠΏΠΎΠ»ΡŒΠ·ΡƒΠ΅Ρ‚ ΠΌΠ΅Ρ‚ΠΎΠ΄Ρ‹ <c>Resolve</c> Π»ΠΈΠ±ΠΎ <c>Reject</c> для оповСщСния ΠΊΠ»ΠΈΠ΅Ρ‚Π½Π° ΠΎ
30 30 /// Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΈΠΈ обСщания.
31 31 /// </para>
32 32 /// <para>
33 33 /// Если сСрвСр успСл Π²Ρ‹ΠΏΠΎΠ»Π½ΠΈΡ‚ΡŒ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ Π΅Ρ‰Π΅ Π΄ΠΎ Ρ‚ΠΎΠ³ΠΎ, ΠΊΠ°ΠΊ ΠΊΠ»ΠΈΠ΅Π½Ρ‚ Π½Π° Π½Π΅Π³ΠΎ подписался,
34 34 /// Ρ‚ΠΎ Π² ΠΌΠΎΠΌΠ΅Π½Ρ‚ подписки ΠΊΠ»ΠΈΠ΅Π½Ρ‚Π° Π±ΡƒΠ΄ΡƒΡ‚ Π²Ρ‹Π·Π²Π°Π½Ρ‹ ΡΠΎΠΎΡ‚Π²Π΅Ρ‚ΡΠ²ΡƒΡŽΡ‰ΠΈΠ΅ события Π² синхронном
35 35 /// Ρ€Π΅ΠΆΠΈΠΌΠ΅ ΠΈ ΠΊΠ»ΠΈΠ΅Π½Ρ‚ Π±ΡƒΠ΄Π΅Ρ‚ ΠΎΠΏΠΎΠ²Π΅Ρ‰Π΅Π½ Π² любом случаС. Π˜Π½Π°Ρ‡Π΅, ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊΠΈ Π΄ΠΎΠ±Π°Π²Π»ΡΡŽΡ‚ΡΡ Π²
36 36 /// список Π² порядкС подписания ΠΈ Π² этом ΠΆΠ΅ порядкС ΠΎΠ½ΠΈ Π±ΡƒΠ΄ΡƒΡ‚ Π²Ρ‹Π·Π²Π°Π½Ρ‹ ΠΏΡ€ΠΈ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΈΠΈ
37 37 /// обСщания.
38 38 /// </para>
39 39 /// <para>
40 40 /// ΠžΠ±Ρ€Π°Π±Π°Ρ‚Ρ‹Π²Π°Ρ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Ρ‹ обСщания ΠΌΠΎΠΆΠ½ΠΎ ΠΏΡ€Π΅ΠΎΠ±Ρ€Π°Π·ΠΎΠ²Ρ‹Π²Π°Ρ‚ΡŒ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Ρ‹ Π»ΠΈΠ±ΠΎ ΠΈΠ½ΠΈΡ†ΠΈΠΈΡ€ΠΎΠ²Π°Ρ‚ΡŒ
41 41 /// связанныС асинхронныС ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ, ΠΊΠΎΡ‚ΠΎΡ€Ρ‹Π΅ Ρ‚Π°ΠΊΠΆΠ΅ Π²ΠΎΠ·Π²Ρ€Π°Ρ‰Π°ΡŽΡ‚ обСщания. Для этого слСдуСт
42 42 /// ΠΈΡΠΏΠΎΠ»ΡŒΠ·ΠΎΠ²Π°Ρ‚ΡŒ ΡΠΎΠΎΡ‚Π²Π΅Ρ‚ΡΡ‚Π²ΡƒΡŽΡ‰ΡƒΡŽ Ρ„ΠΎΡ€ΠΌΡƒ ΠΌΠ΅Ρ‚ΠΎΠ΄Π΅ <c>Then</c>.
43 43 /// </para>
44 44 /// <para>
45 45 /// Π’Π°ΠΊΠΆΠ΅ Ρ…ΠΎΡ€ΠΎΡˆΠΈΠΌ ΠΏΡ€Π°Π²ΠΈΠ»ΠΎΠΌ являСтся Ρ‚ΠΎ, Ρ‡Ρ‚ΠΎ <c>Resolve</c> ΠΈ <c>Reject</c> Π΄ΠΎΠ»ΠΆΠ΅Π½ Π²Ρ‹Π·Ρ‹Π²Π°Ρ‚ΡŒ
46 46 /// Ρ‚ΠΎΠ»ΡŒΠΊΠΎ ΠΈΠ½ΠΈΡ†ΠΈΠ°Ρ‚ΠΎΡ€ обСщания ΠΈΠ½Π°Ρ‡Π΅ ΠΌΠΎΠ³ΡƒΡ‚ Π²ΠΎΠ·Π½ΠΈΠΊΠ½ΡƒΡ‚ΡŒ противорСчия.
47 47 /// </para>
48 48 /// </remarks>
49 49 public class Promise<T> : IPromise {
50 50
51 51 struct ResultHandlerInfo {
52 52 public ResultHandler<T> resultHandler;
53 53 public ErrorHandler errorHandler;
54 54 }
55 55
56 56 readonly IPromise m_parent;
57 57
58 58 LinkedList<ResultHandlerInfo> m_resultHandlers = new LinkedList<ResultHandlerInfo>();
59 59 LinkedList<Action> m_cancelHandlers = new LinkedList<Action>();
60 60
61 61 readonly object m_lock = new Object();
62 62 readonly bool m_cancellable;
63 63 int m_childrenCount = 0;
64 64
65 65 PromiseState m_state;
66 66 T m_result;
67 67 Exception m_error;
68 68
69 69 public Promise() {
70 70 m_cancellable = true;
71 71 }
72 72
73 73 public Promise(IPromise parent, bool cancellable) {
74 74 m_cancellable = cancellable;
75 75 m_parent = parent;
76 76 if (parent != null)
77 77 parent.HandleCancelled(InternalCancel);
78 78 }
79 79
80 80 void InternalCancel() {
81 81 // don't try to cancel parent :)
82 82 Cancel(false);
83 83 }
84 84
85 85 /// <summary>
86 86 /// ВыполняСт ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅, сообщая ΠΎΠ± ΡƒΡΠΏΠ΅ΡˆΠ½ΠΎΠΌ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΈΠΈ.
87 87 /// </summary>
88 88 /// <param name="result">Π Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚ выполнСния.</param>
89 89 /// <exception cref="InvalidOperationException">Π”Π°Π½Π½ΠΎΠ΅ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ ΡƒΠΆΠ΅ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΎ</exception>
90 90 public void Resolve(T result) {
91 91 lock (m_lock) {
92 92 if (m_state == PromiseState.Cancelled)
93 93 return;
94 94 if (m_state != PromiseState.Unresolved)
95 95 throw new InvalidOperationException("The promise is already resolved");
96 96 m_result = result;
97 97 m_state = PromiseState.Resolved;
98 98 }
99 99
100 100 OnStateChanged();
101 101 }
102 102
103 103 /// <summary>
104 104 /// ВыполняСт ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅, сообщая ΠΎΠ± ошибкС
105 105 /// </summary>
106 106 /// <param name="error">Π˜ΡΠΊΠ»ΡŽΡ‡Π΅Π½ΠΈΠ΅ возникшСС ΠΏΡ€ΠΈ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΈΠΈ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ</param>
107 107 /// <exception cref="InvalidOperationException">Π”Π°Π½Π½ΠΎΠ΅ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ ΡƒΠΆΠ΅ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΎ</exception>
108 108 public void Reject(Exception error) {
109 109 lock (m_lock) {
110 110 if (m_state == PromiseState.Cancelled)
111 111 return;
112 112 if (m_state != PromiseState.Unresolved)
113 113 throw new InvalidOperationException("The promise is already resolved");
114 114 m_error = error;
115 115 m_state = PromiseState.Rejected;
116 116 }
117 117
118 118 OnStateChanged();
119 119 }
120 120
121 121 /// <summary>
122 122 /// ΠžΡ‚ΠΌΠ΅Π½ΡΠ΅Ρ‚ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΡŽ, Ссли это Π²ΠΎΠ·ΠΌΠΎΠΆΠ½ΠΎ.
123 123 /// </summary>
124 124 /// <returns><c>true</c> ΠžΠΏΠ΅Ρ€Π°Ρ†ΠΈΡ Π±Ρ‹Π»Π° ΠΎΡ‚ΠΌΠ΅Π½Π΅Π½Π°, ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊΠΈ Π½Π΅ Π±ΡƒΠ΄ΡƒΡ‚ Π²Ρ‹Π·Π²Π°Π½Ρ‹.<c>false</c> ΠΎΡ‚ΠΌΠ΅Π½Π° Π½Π΅ Π²ΠΎΠ·ΠΌΠΎΠΆΠ½Π°, ΠΏΠΎΡΠΊΠΎΠ»ΡŒΠΊΡƒ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ ΡƒΠΆΠ΅ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΎ ΠΈ ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊΠΈ ΠΎΡ‚Ρ€Π°Π±ΠΎΡ‚Π°Π»ΠΈ.</returns>
125 125 public bool Cancel() {
126 126 return Cancel(true);
127 127 }
128 128
129 129 /// <summary>
130 130 /// Adds new handlers to this promise.
131 131 /// </summary>
132 132 /// <param name="success">The handler of the successfully completed operation.
133 133 /// This handler will recieve an operation result as a parameter.</param>
134 134 /// <param name="error">Handles an exception that may occur during the operation.</param>
135 135 /// <returns>The new promise chained to this one.</returns>
136 136 public Promise<T> Then(ResultHandler<T> success, ErrorHandler error) {
137 137 if (success == null && error == null)
138 138 return this;
139 139
140 140 var medium = new Promise<T>(this, true);
141 141
142 142 var handlerInfo = new ResultHandlerInfo();
143 143
144 144 if (success != null)
145 145 handlerInfo.resultHandler = x => {
146 146 success(x);
147 147 medium.Resolve(x);
148 148 };
149 149 else
150 150 handlerInfo.resultHandler = medium.Resolve;
151 151
152 152 if (error != null)
153 153 handlerInfo.errorHandler = x => {
154 154 try {
155 155 error(x);
156 156 } catch { }
157 157 medium.Reject(x);
158 158 };
159 159 else
160 160 handlerInfo.errorHandler = medium.Reject;
161 161
162 162 AddHandler(handlerInfo);
163 163
164 164 return medium;
165 165 }
166 166
167 167 /// <summary>
168 168 /// Adds new handlers to this promise.
169 169 /// </summary>
170 170 /// <param name="success">The handler of the successfully completed operation.
171 171 /// This handler will recieve an operation result as a parameter.</param>
172 172 /// <param name="error">Handles an exception that may occur during the operation and returns the value which will be used as the result of the operation.</param>
173 173 /// <returns>The new promise chained to this one.</returns>
174 174 public Promise<T> Then(ResultHandler<T> success, ErrorHandler<T> error) {
175 175 if (success == null && error == null)
176 176 return this;
177 177
178 178 var medium = new Promise<T>(this, true);
179 179
180 180 var handlerInfo = new ResultHandlerInfo();
181 181
182 182 if (success != null)
183 183 handlerInfo.resultHandler = x => {
184 184 success(x);
185 185 medium.Resolve(x);
186 186 };
187 187 else
188 188 handlerInfo.resultHandler = medium.Resolve;
189 189
190 190 if (error != null)
191 191 handlerInfo.errorHandler = x => {
192 192 try {
193 193 medium.Resolve(error(x));
194 194 } catch { }
195 195 medium.Reject(x);
196 196 };
197 197 else
198 198 handlerInfo.errorHandler = medium.Reject;
199 199
200 200 AddHandler(handlerInfo);
201 201
202 202 return medium;
203 203 }
204 204
205 205
206 206 public Promise<T> Then(ResultHandler<T> success) {
207 207 if (success == null)
208 208 return this;
209 209
210 210 var medium = new Promise<T>(this, true);
211 211
212 212 var handlerInfo = new ResultHandlerInfo();
213 213
214 214 if (success != null)
215 215 handlerInfo.resultHandler = x => {
216 216 success(x);
217 217 medium.Resolve(x);
218 218 };
219 219 else
220 220 handlerInfo.resultHandler = medium.Resolve;
221 221
222 222 handlerInfo.errorHandler = medium.Reject;
223 223
224 224 AddHandler(handlerInfo);
225 225
226 226 return medium;
227 227 }
228 228
229 229 public Promise<T> Error(ErrorHandler error) {
230 230 return Then(null, error);
231 231 }
232 232
233 233 /// <summary>
234 234 /// Handles error and allows to keep the promise.
235 235 /// </summary>
236 236 /// <remarks>
237 237 /// If the specified handler throws an exception, this exception will be used to reject the promise.
238 238 /// </remarks>
239 239 /// <param name="handler">The error handler which returns the result of the promise.</param>
240 240 /// <returns>New promise.</returns>
241 241 public Promise<T> Error(ErrorHandler<T> handler) {
242 242 if (handler == null)
243 243 return this;
244 244
245 245 var medium = new Promise<T>(this, true);
246 246
247 247 AddHandler(new ResultHandlerInfo {
248 248 errorHandler = e => {
249 249 try {
250 250 medium.Resolve(handler(e));
251 251 } catch (Exception e2) {
252 252 medium.Reject(e2);
253 253 }
254 254 }
255 255 });
256 256
257 257 return medium;
258 258 }
259 259
260 260 public Promise<T> Anyway(Action handler) {
261 261 if (handler == null)
262 262 return this;
263 263
264 264 var medium = new Promise<T>();
265 265
266 266 AddHandler(new ResultHandlerInfo {
267 267 resultHandler = x => {
268 268 // to avoid handler being called multiple times we handle exception by ourselfs
269 269 try {
270 270 handler();
271 271 medium.Resolve(x);
272 272 } catch (Exception e) {
273 273 medium.Reject(e);
274 274 }
275 275 },
276 276 errorHandler = x => {
277 277 try {
278 278 handler();
279 279 } catch { }
280 280 medium.Reject(x);
281 281 }
282 282 });
283 283
284 284 return medium;
285 285 }
286 286
287 287 /// <summary>
288 288 /// ΠŸΠΎΠ·Π²ΠΎΠ»ΡΠ΅Ρ‚ ΠΏΡ€Π΅ΠΎΠ±Ρ€Π°Π·ΠΎΠ²Π°Ρ‚ΡŒ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚ выполСния ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ ΠΊ Π½ΠΎΠ²ΠΎΠΌΡƒ Ρ‚ΠΈΠΏΡƒ.
289 289 /// </summary>
290 290 /// <typeparam name="TNew">Новый Ρ‚ΠΈΠΏ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Π°.</typeparam>
291 291 /// <param name="mapper">ΠŸΡ€Π΅ΠΎΠ±Ρ€Π°Π·ΠΎΠ²Π°Π½ΠΈΠ΅ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Π° ΠΊ Π½ΠΎΠ²ΠΎΠΌΡƒ Ρ‚ΠΈΠΏΡƒ.</param>
292 292 /// <param name="error">ΠžΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊ ошибки. Π”Π°Π½Π½Ρ‹ΠΉ ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊ ΠΏΠΎΠ»ΡƒΡ‡ΠΈΡ‚
293 293 /// ΠΈΡΠΊΠ»ΡŽΡ‡Π΅Π½ΠΈΠ΅ возникшСС ΠΏΡ€ΠΈ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΈΠΈ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ.</param>
294 294 /// <returns>НовоС ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅, ΠΊΠΎΡ‚ΠΎΡ€ΠΎΠ΅ Π±ΡƒΠ΄Π΅Ρ‚ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΎ ΠΏΡ€ΠΈ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΈΠΈ исходного обСщания.</returns>
295 295 public Promise<TNew> Map<TNew>(ResultMapper<T, TNew> mapper, ErrorHandler error) {
296 296 if (mapper == null)
297 297 throw new ArgumentNullException("mapper");
298 298
299 299 // создаСм ΠΏΡ€ΠΈΡ†Π΅ΠΏΠ»Π΅Π½Π½ΠΎΠ΅ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅
300 300 var chained = new Promise<TNew>();
301 301
302 302 AddHandler(new ResultHandlerInfo() {
303 303 resultHandler = result => chained.Resolve(mapper(result)),
304 304 errorHandler = delegate(Exception e) {
305 305 if (error != null)
306 306 try {
307 307 error(e);
308 308 } catch { }
309 309 // Π² случаС ошибки Π½ΡƒΠΆΠ½ΠΎ ΠΏΠ΅Ρ€Π΅Π΄Π°Ρ‚ΡŒ ΠΈΡΠΊΠ»ΡŽΡ‡Π΅Π½ΠΈΠ΅ дальшС ΠΏΠΎ Ρ†Π΅ΠΏΠΎΡ‡ΠΊΠ΅
310 310 chained.Reject(e);
311 311 }
312 312 });
313 313
314 314 return chained;
315 315 }
316 316
317 317 public Promise<TNew> Map<TNew>(ResultMapper<T, TNew> mapper) {
318 318 return Map(mapper, null);
319 319 }
320 320
321 321 /// <summary>
322 322 /// БцСпляСт нСсколько аснхронных ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΉ. Указанная асинхронная опСрация Π±ΡƒΠ΄Π΅Ρ‚ Π²Ρ‹Π·Π²Π°Π½Π° послС
323 323 /// выполнСния Ρ‚Π΅ΠΊΡƒΡ‰Π΅ΠΉ, Π° Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚ Ρ‚Π΅ΠΊΡƒΡ‰Π΅ΠΉ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ ΠΌΠΎΠΆΠ΅Ρ‚ Π±Ρ‹Ρ‚ΡŒ использован для ΠΈΠ½ΠΈΡ†ΠΈΠ°Π»ΠΈΠ·Π°Ρ†ΠΈΠΈ
324 324 /// Π½ΠΎΠ²ΠΎΠΉ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ.
325 325 /// </summary>
326 326 /// <typeparam name="TNew">Π’ΠΈΠΏ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Π° ΡƒΠΊΠ°Π·Π°Π½Π½ΠΎΠΉ асинхронной ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ.</typeparam>
327 327 /// <param name="chained">Асинхронная опСрация, которая Π΄ΠΎΠ»ΠΆΠ½Π° Π±ΡƒΠ΄Π΅Ρ‚ Π½Π°Ρ‡Π°Ρ‚ΡŒΡΡ послС выполнСния Ρ‚Π΅ΠΊΡƒΡ‰Π΅ΠΉ.</param>
328 328 /// <param name="error">ΠžΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊ ошибки. Π”Π°Π½Π½Ρ‹ΠΉ ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊ ΠΏΠΎΠ»ΡƒΡ‡ΠΈΡ‚
329 329 /// ΠΈΡΠΊΠ»ΡŽΡ‡Π΅Π½ΠΈΠ΅ возникшСС ΠΏΡ€ΠΈ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΈΠΈ Ρ‚Π΅ΠΊΡƒΠ΅Ρ‰ΠΉ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ.</param>
330 330 /// <returns>НовоС ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅, ΠΊΠΎΡ‚ΠΎΡ€ΠΎΠ΅ Π±ΡƒΠ΄Π΅Ρ‚ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΎ ΠΏΠΎ ΠΎΠΊΠΎΠ½Ρ‡Π°Π½ΠΈΡŽ ΡƒΠΊΠ°Π·Π°Π½Π½ΠΎΠΉ аснхронной ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ.</returns>
331 331 public Promise<TNew> Chain<TNew>(ChainedOperation<T, TNew> chained, ErrorHandler error) {
332 332
333 333 // ΠΏΡ€ΠΎΠ±Π»Π΅ΠΌΠ° Π² Ρ‚ΠΎΠΌ, Ρ‡Ρ‚ΠΎ Π½Π° ΠΌΠΎΠΌΠ΅Π½Ρ‚ связывания Π΅Ρ‰Π΅ Π½Π΅ Π½Π°Ρ‡Π°Ρ‚Π° асинхронная опСрация, поэтому Π½ΡƒΠΆΠ½ΠΎ
334 334 // ΡΠΎΠ·Π΄Π°Ρ‚ΡŒ посрСдника, ΠΊ ΠΊΠΎΡ‚ΠΎΡ€ΠΎΠΌΡƒ Π±ΡƒΠ΄ΡƒΡ‚ ΠΏΠΎΠ΄Π²Ρ‹Π·ΡΠ²Π°Ρ‚ΡŒΡΡ ΡΠ»Π΅Π΄ΡƒΡŽΡ‰ΠΈΠ΅ ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊΠΈ.
335 335 // ΠΊΠΎΠ³Π΄Π° Π±ΡƒΠ΄Π΅Ρ‚ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½Π° Ρ€Π΅Π°Π»ΡŒΠ½Π°Ρ асинхронная опСрация, ΠΎΠ½Π° ΠΎΠ±Ρ€Π°Ρ‚ΠΈΡ‚ΡŒΡΡ ΠΊ посрСднику, Ρ‡Ρ‚ΠΎΠ±Ρ‹
336 336 // ΠΏΠ΅Ρ€Π΅Π΄Π°Ρ‚ΡŒ Ρ‡Π΅Ρ€Π΅Π· Π½Π΅Π³ΠΎ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Ρ‹ Ρ€Π°Π±ΠΎΡ‚Ρ‹.
337 337 var medium = new Promise<TNew>(this, true);
338 338
339 339 AddHandler(new ResultHandlerInfo {
340 340 resultHandler = delegate(T result) {
341 341 if (medium.State == PromiseState.Cancelled)
342 342 return;
343 343
344 344 var promise = chained(result);
345 345
346 346 // notify chained operation that it's not needed
347 347 medium.Cancelled(() => promise.Cancel());
348 348 promise.Then(
349 349 x => medium.Resolve(x),
350 350 e => medium.Reject(e)
351 351 );
352 352 },
353 353 errorHandler = delegate(Exception e) {
354 354 if (error != null)
355 355 error(e);
356 356 // Π² случаС ошибки Π½ΡƒΠΆΠ½ΠΎ ΠΏΠ΅Ρ€Π΅Π΄Π°Ρ‚ΡŒ ΠΈΡΠΊΠ»ΡŽΡ‡Π΅Π½ΠΈΠ΅ дальшС ΠΏΠΎ Ρ†Π΅ΠΏΠΎΡ‡ΠΊΠ΅
357 357 medium.Reject(e);
358 358 }
359 359 });
360 360
361 361 return medium;
362 362 }
363 363
364 364 public Promise<TNew> Chain<TNew>(ChainedOperation<T, TNew> chained) {
365 365 return Chain(chained, null);
366 366 }
367 367
368 368 public Promise<T> Cancelled(Action handler) {
369 369 if (handler == null)
370 370 return this;
371 371 lock (m_lock) {
372 372 if (m_state == PromiseState.Unresolved)
373 373 m_cancelHandlers.AddLast(handler);
374 374 else if (m_state == PromiseState.Cancelled)
375 375 handler();
376 376 }
377 377 return this;
378 378 }
379 379
380 380 public void HandleCancelled(Action handler) {
381 381 Cancelled(handler);
382 382 }
383 383
384 384 /// <summary>
385 385 /// ДоТидаСтся ΠΎΡ‚Π»ΠΎΠΆΠ΅Π½Π½ΠΎΠ³ΠΎ обСщания ΠΈ Π² случаС успСха, Π²ΠΎΠ·Π²Ρ€Π°Ρ‰Π°Π΅Ρ‚
386 386 /// Π΅Π³ΠΎ, Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚, Π² ΠΏΡ€ΠΎΡ‚ΠΈΠ²Π½ΠΎΠΌ случаС бросаСт ΠΈΡΠΊΠ»ΡŽΡ‡Π΅Π½ΠΈΠ΅.
387 387 /// </summary>
388 388 /// <remarks>
389 389 /// <para>
390 390 /// Если ΠΎΠΆΠΈΠ΄Π°Π½ΠΈΠ΅ обСщания Π±Ρ‹Π»ΠΎ ΠΏΡ€Π΅Ρ€Π²Π°Π½ΠΎ ΠΏΠΎ Ρ‚Π°ΠΉΠΌΠ°ΡƒΡ‚Ρƒ, это Π½Π΅ Π·Π½Π°Ρ‡ΠΈΡ‚,
391 391 /// Ρ‡Ρ‚ΠΎ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ Π±Ρ‹Π»ΠΎ ΠΎΡ‚ΠΌΠ΅Π½Π΅Π½ΠΎ ΠΈΠ»ΠΈ Ρ‡Ρ‚ΠΎ-Ρ‚ΠΎ Π² этом Ρ€ΠΎΠ΄Π΅, это Ρ‚ΠΎΠ»ΡŒΠΊΠΎ
392 392 /// ΠΎΠ·Π½Π°Ρ‡Π°Π΅Ρ‚, Ρ‡Ρ‚ΠΎ ΠΌΡ‹ Π΅Π³ΠΎ Π½Π΅ доТдались, ΠΎΠ΄Π½Π°ΠΊΠΎ всС зарСгистрированныС
393 393 /// ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊΠΈ, ΠΊΠ°ΠΊ Π±Ρ‹Π»ΠΈ Ρ‚Π°ΠΊ ΠΎΡΡ‚Π°Π»ΠΈΡΡŒ ΠΈ ΠΎΠ½ΠΈ Π±ΡƒΠ΄ΡƒΡ‚ Π²Ρ‹Π·Π²Π°Π½Ρ‹, ΠΊΠΎΠ³Π΄Π°
394 394 /// ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ Π±ΡƒΠ΄Π΅Ρ‚ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΎ.
395 395 /// </para>
396 396 /// <para>
397 397 /// Π’Π°ΠΊΠΎΠ΅ ΠΏΠΎΠ²Π΅Π΄Π΅Π½ΠΈΠ΅ Π²ΠΏΠΎΠ»Π½Π΅ ΠΎΠΏΡ€Π°Π²Π΄Π°Π½ΠΎ ΠΏΠΎΡΠΊΠΎΠ»ΡŒΠΊΡƒ Ρ‚Π°ΠΉΠΌΠ°ΡƒΡ‚ ΠΌΠΎΠΆΠ΅Ρ‚ ΠΈΡΡ‚Π΅Ρ‡ΡŒ
398 398 /// Π² Ρ‚ΠΎΡ‚ ΠΌΠΎΠΌΠ΅Π½Ρ‚, ΠΊΠΎΠ³Π΄Π° Π½Π°Ρ‡Π°Π»Π°ΡΡŒ ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚ΠΊΠ° Ρ†Π΅ΠΏΠΎΡ‡ΠΊΠΈ ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊΠΎΠ², ΠΈ
399 399 /// ΠΊ Ρ‚ΠΎΠΌΡƒ ΠΆΠ΅ Ρ‚Π΅ΠΊΡƒΡ‰Π΅Π΅ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ ΠΌΠΎΠΆΠ΅Ρ‚ ΡΡ‚ΠΎΡΡ‚ΡŒ Π² Ρ†Π΅ΠΏΠΎΡ‡ΠΊΠ΅ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠΉ ΠΈ Π΅Π³ΠΎ
400 400 /// ΠΎΡ‚ΠΊΠ»ΠΎΠ½Π΅Π½ΠΈΠ΅ ΠΌΠΎΠΆΠ΅Ρ‚ привСсти ΠΊ Π½Π΅ΠΏΡ€ΠΎΠ³Π½ΠΎΠ·ΠΈΡ€ΡƒΠ΅ΠΌΠΎΠΌΡƒ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Ρƒ.
401 401 /// </para>
402 402 /// </remarks>
403 403 /// <param name="timeout">ВрСмя оТидания</param>
404 404 /// <returns>Π Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚ выполнСния обСщания</returns>
405 405 public T Join(int timeout) {
406 406 var evt = new ManualResetEvent(false);
407 407 Anyway(() => evt.Set());
408 408 Cancelled(() => evt.Set());
409 409
410 410 if (!evt.WaitOne(timeout, true))
411 411 throw new TimeoutException();
412 412
413 413 switch (State) {
414 414 case PromiseState.Resolved:
415 415 return m_result;
416 416 case PromiseState.Cancelled:
417 417 throw new OperationCanceledException();
418 418 case PromiseState.Rejected:
419 419 throw new TargetInvocationException(m_error);
420 420 default:
421 421 throw new ApplicationException(String.Format("Invalid promise state {0}", State));
422 422 }
423 423 }
424 424
425 425 public T Join() {
426 426 return Join(Timeout.Infinite);
427 427 }
428 428
429 429 void AddHandler(ResultHandlerInfo handler) {
430 430 bool invokeRequired = false;
431 431
432 432 lock (m_lock) {
433 433 m_childrenCount++;
434 434 if (m_state == PromiseState.Unresolved) {
435 435 m_resultHandlers.AddLast(handler);
436 436 } else
437 437 invokeRequired = true;
438 438 }
439 439
440 440 // ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊΠΈ Π½Π΅ Π΄ΠΎΠ»ΠΆΠ½Ρ‹ Π±Π»ΠΎΠΊΠΈΡ€ΠΎΠ²Π°Ρ‚ΡŒ сам ΠΎΠ±ΡŠΠ΅ΠΊΡ‚
441 441 if (invokeRequired)
442 442 InvokeHandler(handler);
443 443 }
444 444
445 445 void InvokeHandler(ResultHandlerInfo handler) {
446 446 switch (m_state) {
447 447 case PromiseState.Resolved:
448 448 try {
449 449 if (handler.resultHandler != null)
450 450 handler.resultHandler(m_result);
451 451 } catch (Exception e) {
452 452 try {
453 453 if (handler.errorHandler != null)
454 454 handler.errorHandler(e);
455 455 } catch { }
456 456 }
457 457 break;
458 458 case PromiseState.Rejected:
459 459 try {
460 460 if (handler.errorHandler != null)
461 461 handler.errorHandler(m_error);
462 462 } catch { }
463 463 break;
464 464 default:
465 465 // do nothing
466 466 return;
467 467 }
468 468 }
469 469
470 470 protected virtual void OnStateChanged() {
471 471 switch (m_state) {
472 472 case PromiseState.Resolved:
473 473 foreach (var resultHandlerInfo in m_resultHandlers)
474 474 try {
475 475 if (resultHandlerInfo.resultHandler != null)
476 476 resultHandlerInfo.resultHandler(m_result);
477 477 } catch (Exception e) {
478 478 try {
479 479 if (resultHandlerInfo.errorHandler != null)
480 480 resultHandlerInfo.errorHandler(e);
481 481 } catch { }
482 482 }
483 483 break;
484 484 case PromiseState.Cancelled:
485 485 foreach (var cancelHandler in m_cancelHandlers)
486 486 cancelHandler();
487 487 break;
488 488 case PromiseState.Rejected:
489 489 foreach (var resultHandlerInfo in m_resultHandlers)
490 490 try {
491 491 if (resultHandlerInfo.errorHandler != null)
492 492 resultHandlerInfo.errorHandler(m_error);
493 493 } catch { }
494 494 break;
495 495 default:
496 496 throw new InvalidOperationException(String.Format("Promise entered an invalid state {0}", m_state));
497 497 }
498 498
499 499 m_resultHandlers = null;
500 500 m_cancelHandlers = null;
501 501 }
502 502
503 503
504 504
505 505 public bool IsExclusive {
506 506 get {
507 507 lock (m_lock) {
508 508 return m_childrenCount <= 1;
509 509 }
510 510 }
511 511 }
512 512
513 513 public PromiseState State {
514 514 get {
515 515 lock (m_lock) {
516 516 return m_state;
517 517 }
518 518 }
519 519 }
520 520
521 521 protected bool Cancel(bool dependencies) {
522 522 bool result;
523 523
524 524 lock (m_lock) {
525 525 if (m_state == PromiseState.Unresolved) {
526 526 m_state = PromiseState.Cancelled;
527 527 result = true;
528 528 } else {
529 529 result = false;
530 530 }
531 531 }
532 532
533 533 if (result)
534 534 OnStateChanged();
535 535
536 536 if (dependencies && m_parent != null && m_parent.IsExclusive) {
537 537 m_parent.Cancel();
538 538 }
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