##// END OF EJS Templates
implab » ImplabNet
RSS
sync
cin -
r16:5a4b735ba669 promises
parent child
Show More
Show file before | Show file after | Hide comments
@@ -1,301 +1,333
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 7 namespace Implab.Test {
8 8 [TestClass]
9 9 public class AsyncTests {
10 10 [TestMethod]
11 11 public void ResolveTest() {
12 12 int res = -1;
13 13 var p = new Promise<int>();
14 14 p.Then(x => res = x);
15 15 p.Resolve(100);
16 16
17 17 Assert.AreEqual(res, 100);
18 18 }
19 19
20 20 [TestMethod]
21 21 public void RejectTest() {
22 22 int res = -1;
23 23 Exception err = null;
24 24
25 25 var p = new Promise<int>();
26 26 p.Then(x => res = x, e => err = e);
27 27 p.Reject(new ApplicationException("error"));
28 28
29 29 Assert.AreEqual(res, -1);
30 30 Assert.AreEqual(err.Message, "error");
31 31
32 32 }
33 33
34 34 [TestMethod]
35 35 public void JoinSuccessTest() {
36 36 var p = new Promise<int>();
37 37 p.Resolve(100);
38 38 Assert.AreEqual(p.Join(), 100);
39 39 }
40 40
41 41 [TestMethod]
42 42 public void JoinFailTest() {
43 43 var p = new Promise<int>();
44 44 p.Reject(new ApplicationException("failed"));
45 45
46 46 try {
47 47 p.Join();
48 48 throw new ApplicationException("WRONG!");
49 49 } catch (TargetInvocationException err) {
50 50 Assert.AreEqual(err.InnerException.Message, "failed");
51 51 } catch {
52 52 Assert.Fail("Got wrong excaption");
53 53 }
54 54 }
55 55
56 56 [TestMethod]
57 57 public void MapTest() {
58 58 var p = new Promise<int>();
59 59
60 60 var p2 = p.Map(x => x.ToString());
61 61 p.Resolve(100);
62 62
63 63 Assert.AreEqual(p2.Join(), "100");
64 64 }
65 65
66 66 [TestMethod]
67 67 public void FixErrorTest() {
68 68 var p = new Promise<int>();
69 69
70 70 var p2 = p.Error(e => 101);
71 71
72 72 p.Reject(new Exception());
73 73
74 74 Assert.AreEqual(p2.Join(), 101);
75 75 }
76 76
77 77 [TestMethod]
78 78 public void ChainTest() {
79 79 var p1 = new Promise<int>();
80 80
81 81 var p3 = p1.Chain(x => {
82 82 var p2 = new Promise<string>();
83 83 p2.Resolve(x.ToString());
84 84 return p2;
85 85 });
86 86
87 87 p1.Resolve(100);
88 88
89 89 Assert.AreEqual(p3.Join(), "100");
90 90 }
91 91
92 92 [TestMethod]
93 93 public void PoolTest() {
94 94 var pid = Thread.CurrentThread.ManagedThreadId;
95 95 var p = AsyncPool.Invoke(() => Thread.CurrentThread.ManagedThreadId);
96 96
97 97 Assert.AreNotEqual(pid, p.Join());
98 98 }
99 99
100 100 [TestMethod]
101 101 public void WorkerPoolSizeTest() {
102 102 var pool = new WorkerPool(5, 10);
103 103
104 104 Assert.AreEqual(5, pool.ThreadCount);
105 105
106 106 pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
107 107 pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
108 108 pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
109 109
110 110 Assert.AreEqual(5, pool.ThreadCount);
111 111
112 112 for (int i = 0; i < 100; i++)
113 113 pool.Invoke(() => { Thread.Sleep(1000000); return 10; });
114 114 Thread.Sleep(100);
115 115 Assert.AreEqual(10, pool.ThreadCount);
116 116
117 117 pool.Dispose();
118 118 }
119 119
120 120 [TestMethod]
121 121 public void WorkerPoolCorrectTest() {
122 122 var pool = new WorkerPool();
123 123
124 124 int iterations = 1000;
125 125 int pending = iterations;
126 126 var stop = new ManualResetEvent(false);
127 127
128 128 var count = 0;
129 129 for (int i = 0; i < iterations; i++) {
130 130 pool
131 131 .Invoke(() => 1)
132 132 .Then(x => Interlocked.Add(ref count, x))
133 133 .Then(x => Math.Log10(x))
134 134 .Anyway(() => {
135 135 Interlocked.Decrement(ref pending);
136 136 if (pending == 0)
137 137 stop.Set();
138 138 });
139 139 }
140 140
141 141 stop.WaitOne();
142 142
143 143 Assert.AreEqual(iterations, count);
144 144 Console.WriteLine("Max threads: {0}", pool.MaxRunningThreads);
145 145 pool.Dispose();
146 146
147 147 }
148 148
149 149 [TestMethod]
150 150 public void WorkerPoolDisposeTest() {
151 151 var pool = new WorkerPool(5, 20);
152 152 Assert.AreEqual(5, pool.ThreadCount);
153 153 pool.Dispose();
154 154 Thread.Sleep(100);
155 155 Assert.AreEqual(0, pool.ThreadCount);
156 156 pool.Dispose();
157 157 }
158 158
159 159 [TestMethod]
160 160 public void MTQueueTest() {
161 161 var queue = new MTQueue<int>();
162 162 int res;
163 163
164 164 queue.Enqueue(10);
165 165 Assert.IsTrue(queue.TryDequeue(out res));
166 166 Assert.AreEqual(10, res);
167 167 Assert.IsFalse(queue.TryDequeue(out res));
168 168
169 169 for (int i = 0; i < 1000; i++)
170 170 queue.Enqueue(i);
171 171
172 172 for (int i = 0; i < 1000; i++) {
173 173 queue.TryDequeue(out res);
174 174 Assert.AreEqual(i, res);
175 175 }
176 176
177 177 int writers = 0;
178 178 int readers = 0;
179 179 var stop = new ManualResetEvent(false);
180 180 int total = 0;
181 181
182 182 int itemsPerWriter = 1000;
183 183 int writersCount = 3;
184 184
185 185 for (int i = 0; i < writersCount; i++) {
186 186 Interlocked.Increment(ref writers);
187 187 var wn = i;
188 188 AsyncPool
189 189 .InvokeNewThread(() => {
190 190 for (int ii = 0; ii < itemsPerWriter; ii++) {
191 191 queue.Enqueue(1);
192 192 }
193 193 return 1;
194 194 })
195 195 .Anyway(() => Interlocked.Decrement(ref writers));
196 196 }
197 197
198 198 for (int i = 0; i < 10; i++) {
199 199 Interlocked.Increment(ref readers);
200 200 var wn = i;
201 201 AsyncPool
202 202 .InvokeNewThread(() => {
203 203 int t;
204 204 do {
205 205 while (queue.TryDequeue(out t))
206 206 Interlocked.Add(ref total, t);
207 207 } while (writers > 0);
208 208 return 1;
209 209 })
210 210 .Anyway(() => {
211 211 Interlocked.Decrement(ref readers);
212 212 if (readers == 0)
213 213 stop.Set();
214 214 });
215 215 }
216 216
217 217 stop.WaitOne();
218 218
219 219 Assert.AreEqual(itemsPerWriter * writersCount, total);
220 220 }
221 221
222 222 [TestMethod]
223 223 public void ParallelMapTest() {
224 224
225 225 int count = 100000;
226 226
227 227 double[] args = new double[count];
228 228 var rand = new Random();
229 229
230 230 for (int i = 0; i < count; i++)
231 231 args[i] = rand.NextDouble();
232 232
233 233 var t = Environment.TickCount;
234 234 var res = args.ParallelMap(x => Math.Sin(x*x), 4).Join();
235 235
236 236 Console.WriteLine("Map complete in {0} ms", Environment.TickCount - t);
237 237
238 238 t = Environment.TickCount;
239 239 for (int i = 0; i < count; i++)
240 240 Assert.AreEqual(Math.Sin(args[i] * args[i]), res[i]);
241 241 Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
242 242 }
243 243
244 244 [TestMethod]
245 public void ChainedMapTest() {
246
247 using (var pool = new WorkerPool(1,10)) {
248 int count = 10000;
249
250 double[] args = new double[count];
251 var rand = new Random();
252
253 for (int i = 0; i < count; i++)
254 args[i] = rand.NextDouble();
255
256 var t = Environment.TickCount;
257 var res = args
258 .ChainedMap(
259 x => pool.Invoke(
260 () => Math.Sin(x * x)
261 ),
262 4
263 )
264 .Join();
265
266 Console.WriteLine("Map complete in {0} ms", Environment.TickCount - t);
267
268 t = Environment.TickCount;
269 for (int i = 0; i < count; i++)
270 Assert.AreEqual(Math.Sin(args[i] * args[i]), res[i]);
271 Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
272 Console.WriteLine("Max workers: {0}", pool.MaxRunningThreads);
273 }
274 }
275
276 [TestMethod]
245 277 public void ParallelForEachTest() {
246 278
247 279 int count = 100000;
248 280
249 281 int[] args = new int[count];
250 282 var rand = new Random();
251 283
252 284 for (int i = 0; i < count; i++)
253 285 args[i] = (int)(rand.NextDouble() * 100);
254 286
255 287 int result = 0;
256 288
257 289 var t = Environment.TickCount;
258 290 args.ParallelForEach(x => Interlocked.Add(ref result, x), 4).Join();
259 291
260 292 Console.WriteLine("Iteration complete in {0} ms, result: {1}", Environment.TickCount - t, result);
261 293
262 294 int result2 = 0;
263 295
264 296 t = Environment.TickCount;
265 297 for (int i = 0; i < count; i++)
266 298 result2 += args[i];
267 299 Assert.AreEqual(result2, result);
268 300 Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
269 301 }
270 302
271 303 [TestMethod]
272 304 public void ComplexCase1Test() {
273 305 var flags = new bool[3];
274 306
275 307 // op1 (aync 200ms) => op2 (async 200ms) => op3 (sync map)
276 308
277 309 var p = PromiseHelper
278 310 .Sleep(200, "Alan")
279 311 .Cancelled(() => flags[0] = true)
280 312 .Chain(x =>
281 313 PromiseHelper
282 314 .Sleep(200, "Hi, " + x)
283 315 .Map(y => y)
284 316 .Cancelled(() => flags[1] = true)
285 317 )
286 318 .Cancelled(() => flags[2] = true);
287 319 Thread.Sleep(300);
288 320 p.Cancel();
289 321 try {
290 322 Assert.AreEqual(p.Join(), "Hi, Alan");
291 323 Assert.Fail("Shouldn't get here");
292 324 } catch (OperationCanceledException) {
293 325 }
294 326
295 327 Assert.IsFalse(flags[0]);
296 328 Assert.IsTrue(flags[1]);
297 329 Assert.IsTrue(flags[2]);
298 330 }
299 331 }
300 332 }
301 333
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,152 +1,171
1 1 using System;
2 2 using System.Collections.Generic;
3 3 using System.Diagnostics;
4 4 using System.Linq;
5 5 using System.Text;
6 6 using System.Threading;
7 7
8 8 namespace Implab.Parallels {
9 9 public static class ArrayTraits {
10 10 class ArrayIterator<TSrc> : DispatchPool<int> {
11 11 readonly Action<TSrc> m_action;
12 12 readonly TSrc[] m_source;
13 13 readonly Promise<int> m_promise = new Promise<int>();
14 14
15 15 int m_pending;
16 16 int m_next;
17 17
18 18 public ArrayIterator(TSrc[] source, Action<TSrc> action, int threads)
19 19 : base(threads) {
20 20
21 21 Debug.Assert(source != null);
22 22 Debug.Assert(action != null);
23 23
24 24 m_next = 0;
25 25 m_source = source;
26 26 m_pending = source.Length;
27 27 m_action = action;
28 28
29 29 m_promise.Anyway(() => Dispose());
30 30 m_promise.Cancelled(() => Dispose());
31 31
32 32 InitPool();
33 33 }
34 34
35 35 public Promise<int> Promise {
36 36 get {
37 37 return m_promise;
38 38 }
39 39 }
40 40
41 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;
42 unit = Interlocked.Increment(ref m_next) - 1;
43 return unit >= m_source.Length ? false : true;
52 44 }
53 45
54 46 protected override void InvokeUnit(int unit) {
55 47 try {
56 48 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--;
49 var pending = Interlocked.Decrement(ref m_pending);
62 50 if (pending == 0)
63 51 m_promise.Resolve(m_source.Length);
64 52 } catch (Exception e) {
65 53 m_promise.Reject(e);
66 54 }
67 55 }
68 56 }
69 57
70 58 class ArrayMapper<TSrc, TDst>: DispatchPool<int> {
71 59 readonly Func<TSrc, TDst> m_transform;
72 60 readonly TSrc[] m_source;
73 61 readonly TDst[] m_dest;
74 62 readonly Promise<TDst[]> m_promise = new Promise<TDst[]>();
75 63
76 64 int m_pending;
77 65 int m_next;
78 66
79 67 public ArrayMapper(TSrc[] source, Func<TSrc, TDst> transform, int threads)
80 68 : base(threads) {
81 69
82 70 Debug.Assert (source != null);
83 71 Debug.Assert( transform != null);
84 72
85 73 m_next = 0;
86 74 m_source = source;
87 75 m_dest = new TDst[source.Length];
88 76 m_pending = source.Length;
89 77 m_transform = transform;
90 78
91 79 m_promise.Anyway(() => Dispose());
92 80 m_promise.Cancelled(() => Dispose());
93 81
94 82 InitPool();
95 83 }
96 84
97 85 public Promise<TDst[]> Promise {
98 86 get {
99 87 return m_promise;
100 88 }
101 89 }
102 90
103 91 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;
92 unit = Interlocked.Increment(ref m_next) - 1;
93 return unit >= m_source.Length ? false : true;
114 94 }
115 95
116 96 protected override void InvokeUnit(int unit) {
117 97 try {
118 98 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 --;
99 var pending = Interlocked.Decrement(ref m_pending);
124 100 if (pending == 0)
125 101 m_promise.Resolve(m_dest);
126 102 } catch (Exception e) {
127 103 m_promise.Reject(e);
128 104 }
129 105 }
130 106 }
131 107
132 108 public static Promise<TDst[]> ParallelMap<TSrc, TDst> (this TSrc[] source, Func<TSrc,TDst> transform, int threads) {
133 109 if (source == null)
134 110 throw new ArgumentNullException("source");
135 111 if (transform == null)
136 112 throw new ArgumentNullException("transform");
137 113
138 114 var mapper = new ArrayMapper<TSrc, TDst>(source, transform, threads);
139 115 return mapper.Promise;
140 116 }
141 117
142 118 public static Promise<int> ParallelForEach<TSrc>(this TSrc[] source, Action<TSrc> action, int threads) {
143 119 if (source == null)
144 120 throw new ArgumentNullException("source");
145 121 if (action == null)
146 122 throw new ArgumentNullException("action");
147 123
148 124 var iter = new ArrayIterator<TSrc>(source, action, threads);
149 125 return iter.Promise;
150 126 }
127
128 public static Promise<TDst[]> ChainedMap<TSrc, TDst>(this TSrc[] source, ChainedOperation<TSrc, TDst> transform, int threads) {
129 if (source == null)
130 throw new ArgumentNullException("source");
131 if (transform == null)
132 throw new ArgumentNullException("transform");
133 if (threads <= 0)
134 throw new ArgumentOutOfRangeException("Threads number must be greater then zero");
135
136 var promise = new Promise<TDst[]>();
137 var res = new TDst[source.Length];
138 var pending = source.Length;
139 var semaphore = new Semaphore(threads, threads);
140
141 AsyncPool.InvokeNewThread(() => {
142 for (int i = 0; i < source.Length; i++) {
143 if(promise.State != PromiseState.Unresolved)
144 break; // stop processing in case of error or cancellation
145 var idx = i;
146 semaphore.WaitOne();
147 try {
148 var p1 = transform(source[i]);
149 p1.Anyway(() => semaphore.Release());
150 p1.Cancelled(() => semaphore.Release());
151 p1.Then(
152 x => {
153 res[idx] = x;
154 var left = Interlocked.Decrement(ref pending);
155 if (left == 0)
156 promise.Resolve(res);
157 },
158 e => promise.Reject(e)
159 );
160
161 } catch (Exception e) {
162 promise.Reject(e);
151 163 }
152 164 }
165 return 0;
166 });
167
168 return promise.Anyway(() => semaphore.Dispose());
169 }
170 }
171 }
Show file before | Show file after | Hide comments
@@ -1,45 +1,44
1 1 using System;
2 2 using System.Threading;
3 3
4 4 namespace Implab.Parallels {
5 5 /// <summary>
6 6 /// Класс для распаралСливания Π·Π°Π΄Π°Ρ‡.
7 7 /// </summary>
8 8 /// <remarks>
9 9 /// Π˜ΡΠΏΠΎΠ»ΡŒΠ·ΡƒΡ Π΄Π°Π½Π½Ρ‹ΠΉ класс ΠΈ лямда выраТСния ΠΌΠΎΠΆΠ½ΠΎ Ρ€Π°ΡΠΏΠ°Ρ€Π°Π»Π»Π΅Π»ΠΈΡ‚ΡŒ
10 10 /// вычислСния, для этого ΠΈΡΠΏΠΎΠ»ΡŒΠ·ΡƒΠ΅Ρ‚ΡΡ концСпция ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠΉ.
11 11 /// </remarks>
12 12 public static class AsyncPool {
13 13
14 14 public static Promise<T> Invoke<T>(Func<T> func) {
15 15 var p = new Promise<T>();
16 16
17 17 ThreadPool.QueueUserWorkItem(param => {
18 18 try {
19 19 p.Resolve(func());
20 20 } catch(Exception e) {
21 21 p.Reject(e);
22 22 }
23 23 });
24 24
25 25 return p;
26 26 }
27 27
28 28 public static Promise<T> InvokeNewThread<T>(Func<T> func) {
29 29 var p = new Promise<T>();
30 30
31 31 var worker = new Thread(() => {
32 32 try {
33 33 p.Resolve(func());
34 34 } catch (Exception e) {
35 35 p.Reject(e);
36 36 }
37 37 });
38 38 worker.IsBackground = true;
39
40 39 worker.Start();
41 40
42 41 return p;
43 42 }
44 43 }
45 44 }
Show file before | Show file after | Hide comments
@@ -1,171 +1,183
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 9 public abstract class DispatchPool<TUnit> : IDisposable {
10 10 readonly int m_minThreads;
11 11 readonly int m_maxThreads;
12 12 int m_runningThreads = 0;
13 13 int m_maxRunningThreads = 0;
14 14 int m_suspended = 0;
15 15 int m_exitRequired = 0;
16 16 AutoResetEvent m_hasTasks = new AutoResetEvent(false);
17 17
18 18 protected DispatchPool(int min, int max) {
19 19 if (min < 0)
20 20 throw new ArgumentOutOfRangeException("min");
21 21 if (max <= 0)
22 22 throw new ArgumentOutOfRangeException("max");
23 23
24 24 if (min > max)
25 25 min = max;
26 26 m_minThreads = min;
27 27 m_maxThreads = max;
28 28 }
29 29
30 30 protected DispatchPool(int threads)
31 31 : this(threads, threads) {
32 32 }
33 33
34 34 protected DispatchPool() {
35 35 int maxThreads, maxCP;
36 36 ThreadPool.GetMaxThreads(out maxThreads, out maxCP);
37 37
38 38 m_minThreads = 0;
39 39 m_maxThreads = maxThreads;
40 40 }
41 41
42 42 protected void InitPool() {
43 43 for (int i = 0; i < m_minThreads; i++)
44 44 StartWorker();
45 45 }
46 46
47 47 public int ThreadCount {
48 48 get {
49 49 return m_runningThreads;
50 50 }
51 51 }
52 52
53 53 public int MaxRunningThreads {
54 54 get {
55 55 return m_maxRunningThreads;
56 56 }
57 57 }
58 58
59 59 protected bool IsDisposed {
60 60 get {
61 61 return m_exitRequired != 0;
62 62 }
63 63 }
64 64
65 65 bool StartWorker() {
66 var current = m_runningThreads;
66 int current;
67 67 // use spins to allocate slot for the new thread
68 68 do {
69 current = m_runningThreads;
69 70 if (current >= m_maxThreads || m_exitRequired != 0)
70 71 // no more slots left or the pool has been disposed
71 72 return false;
72 73 } while (current != Interlocked.CompareExchange(ref m_runningThreads, current + 1, current));
73 74
74 75 m_maxRunningThreads = Math.Max(m_maxRunningThreads, current + 1);
75 76
76 77 // slot successfully allocated
77 78
78 79 var worker = new Thread(this.Worker);
79 80 worker.IsBackground = true;
80 81 worker.Start();
81 82
82 83 return true;
83 84 }
84 85
85 86 protected abstract bool TryDequeue(out TUnit unit);
86 87
87 protected virtual void WakeNewWorker() {
88 protected virtual void WakeNewWorker(bool extend) {
88 89 if (m_suspended > 0)
89 90 m_hasTasks.Set();
90 91 else
91 92 StartWorker();
92 93 }
93 94
95 /// <summary>
96 /// ЗапускаСт Π»ΠΈΠ±ΠΎ Π½ΠΎΠ²Ρ‹ΠΉ ΠΏΠΎΡ‚ΠΎΠΊ, Ссли Ρ€Π°Π½ΡŒΡˆΠ΅ Π½Π΅ Π±Ρ‹Π»ΠΎ Π½ΠΈ ΠΎΠ΄Π½ΠΎΠ³ΠΎ ΠΏΠΎΡ‚ΠΎΠΊΠ°, Π»ΠΈΠ±ΠΎ устанавливаСт событиС ΠΏΡ€ΠΎΠ±ΡƒΠΆΠ΄Π΅Π½ΠΈΠ΅ ΠΎΠ΄Π½ΠΎΠ³ΠΎ спящСго ΠΏΠΎΡ‚ΠΎΠΊΠ°
97 /// </summary>
98 protected void StartIfIdle() {
99 int threads;
100 do {
101
102 }
103 }
104
105 protected virtual void Suspend() {
106 m_hasTasks.WaitOne();
107 }
108
94 109 bool FetchTask(out TUnit unit) {
95 110 do {
96 111 // exit if requested
97 112 if (m_exitRequired != 0) {
98 113 // 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
114 var running = Interlocked.Decrement(ref m_runningThreads);
105 115 if (running == 0) // it was the last worker
106 116 m_hasTasks.Dispose();
107 117 else
108 118 m_hasTasks.Set(); // release next worker
109 119 unit = default(TUnit);
110 120 return false;
111 121 }
112 122
113 123 // fetch task
114 124 if (TryDequeue(out unit)) {
115 WakeNewWorker();
125 WakeNewWorker(true);
116 126 return true;
117 127 }
118 128
119 129 //no tasks left, exit if the thread is no longer needed
120 130 int runningThreads;
121 131 bool exit = true;
122 132 do {
123 133 runningThreads = m_runningThreads;
124 134 if (runningThreads <= m_minThreads) {
135 // check wheather this is the last thread and we have tasks
136
125 137 exit = false;
126 138 break;
127 139 }
128 140 } while (runningThreads != Interlocked.CompareExchange(ref m_runningThreads, runningThreads - 1, runningThreads));
129 141
130 142 if (exit) {
131 Interlocked.Decrement(ref m_runningThreads);
132 143 return false;
133 144 }
134 145
135 // keep this thread and wait
146 // entering suspend state
136 147 Interlocked.Increment(ref m_suspended);
137 m_hasTasks.WaitOne();
148 // keep this thread and wait
149 Suspend();
138 150 Interlocked.Decrement(ref m_suspended);
139 151 } while (true);
140 152 }
141 153
142 154 protected abstract void InvokeUnit(TUnit unit);
143 155
144 156 void Worker() {
145 157 TUnit unit;
146 158 while (FetchTask(out unit))
147 159 InvokeUnit(unit);
148 160 }
149 161
150 162 protected virtual void Dispose(bool disposing) {
151 163 if (disposing) {
152 164 if (m_exitRequired == 0) {
153 165 if (Interlocked.CompareExchange(ref m_exitRequired, 1, 0) != 0)
154 166 return;
155 167
156 168 // wake sleeping threads
157 169 m_hasTasks.Set();
158 170 GC.SuppressFinalize(this);
159 171 }
160 172 }
161 173 }
162 174
163 175 public void Dispose() {
164 176 Dispose(true);
165 177 }
166 178
167 179 ~DispatchPool() {
168 180 Dispose(false);
169 181 }
170 182 }
171 183 }
Show file before | Show file after | Hide comments
@@ -1,69 +1,90
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 9 public class WorkerPool : DispatchPool<Action> {
10 10
11 11 MTQueue<Action> m_queue = new MTQueue<Action>();
12 12 int m_queueLength = 0;
13 readonly int m_threshold = 1;
13 14
14 public WorkerPool(int minThreads, int maxThreads)
15 public WorkerPool(int minThreads, int maxThreads, int threshold)
15 16 : base(minThreads, maxThreads) {
17 m_threshold = threshold;
18 InitPool();
19 }
20
21 public WorkerPool(int minThreads, int maxThreads) :
22 base(minThreads, maxThreads) {
16 23 InitPool();
17 24 }
18 25
19 26 public WorkerPool(int threads)
20 27 : base(threads) {
21 28 InitPool();
22 29 }
23 30
24 31 public WorkerPool()
25 32 : base() {
26 33 InitPool();
27 34 }
28 35
29 36 public Promise<T> Invoke<T>(Func<T> task) {
30 37 if (task == null)
31 38 throw new ArgumentNullException("task");
32 39 if (IsDisposed)
33 40 throw new ObjectDisposedException(ToString());
34 41
35 42 var promise = new Promise<T>();
36 43
37 44 EnqueueTask(delegate() {
38 45 try {
39 46 promise.Resolve(task());
40 47 } catch (Exception e) {
41 48 promise.Reject(e);
42 49 }
43 50 });
44 51
45 52 return promise;
46 53 }
47 54
48 55 protected void EnqueueTask(Action unit) {
49 56 Debug.Assert(unit != null);
50 Interlocked.Increment(ref m_queueLength);
57 var len = Interlocked.Increment(ref m_queueLength);
51 58 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();
59
60 if (ThreadCount == 0)
61 // force to start
62 WakeNewWorker(false);
63 }
64
65 protected override void WakeNewWorker(bool extend) {
66 if (extend && m_queueLength <= m_threshold)
67 // in this case we are in active thread and it request for additional workers
68 // satisfy it only when queue is longer than threshold
69 return;
70 base.WakeNewWorker(extend);
55 71 }
56 72
57 73 protected override bool TryDequeue(out Action unit) {
58 74 if (m_queue.TryDequeue(out unit)) {
59 75 Interlocked.Decrement(ref m_queueLength);
60 76 return true;
61 77 }
62 78 return false;
63 79 }
64 80
65 81 protected override void InvokeUnit(Action unit) {
66 82 unit();
67 83 }
84
85 protected override void Suspend() {
86 if (m_queueLength == 0)
87 base.Suspend();
68 88 }
69 89 }
90 }
Show file before | Show file after | Hide comments
@@ -1,544 +1,549
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 /// <remarks>
107 /// ΠŸΠΎΡΠΊΠΎΠ»ΡŒΠΊΡƒ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ Π΄ΠΎΠ»ΠΆΠ½ΠΎ Ρ€Π°Π±ΠΎΡ‚Π°Ρ‚ΡŒ Π² ΠΌΠ½ΠΎΠ³ΠΎΠΏΡ‚ΠΎΡ‡Π½ΠΎΠΉ срСдС, ΠΏΡ€ΠΈ Π΅Π³ΠΎ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΈΠΈ сразу нСсколько ΠΏΠΎΡ‚ΠΎΠΊΠΎΠ²
108 /// ΠΌΠΎΠ³Ρƒ Π²Π΅Ρ€Π½ΡƒΡ‚ΡŒ ΠΎΡˆΠΈΠ±ΠΊΡƒ, ΠΏΡ€ΠΈ этом Ρ‚ΠΎΠ»ΡŒΠΊΠΎ пСрвая Π±ΡƒΠ΄Π΅Ρ‚ использована Π² качСствС Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Π°, ΠΎΡΡ‚Π°Π»ΡŒΠ½Ρ‹Π΅
109 /// Π±ΡƒΠ΄ΡƒΡ‚ ΠΏΡ€ΠΎΠΈΠ³Π½ΠΎΡ€ΠΈΡ€ΠΎΠ²Π°Π½Ρ‹.
110 /// </remarks>
106 111 /// <param name="error">Π˜ΡΠΊΠ»ΡŽΡ‡Π΅Π½ΠΈΠ΅ возникшСС ΠΏΡ€ΠΈ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΈΠΈ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ</param>
107 112 /// <exception cref="InvalidOperationException">Π”Π°Π½Π½ΠΎΠ΅ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ ΡƒΠΆΠ΅ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΎ</exception>
108 113 public void Reject(Exception error) {
109 114 lock (m_lock) {
110 if (m_state == PromiseState.Cancelled)
115 if (m_state == PromiseState.Cancelled || m_state == PromiseState.Rejected)
111 116 return;
112 117 if (m_state != PromiseState.Unresolved)
113 118 throw new InvalidOperationException("The promise is already resolved");
114 119 m_error = error;
115 120 m_state = PromiseState.Rejected;
116 121 }
117 122
118 123 OnStateChanged();
119 124 }
120 125
121 126 /// <summary>
122 127 /// ΠžΡ‚ΠΌΠ΅Π½ΡΠ΅Ρ‚ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΡŽ, Ссли это Π²ΠΎΠ·ΠΌΠΎΠΆΠ½ΠΎ.
123 128 /// </summary>
124 129 /// <returns><c>true</c> ΠžΠΏΠ΅Ρ€Π°Ρ†ΠΈΡ Π±Ρ‹Π»Π° ΠΎΡ‚ΠΌΠ΅Π½Π΅Π½Π°, ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊΠΈ Π½Π΅ Π±ΡƒΠ΄ΡƒΡ‚ Π²Ρ‹Π·Π²Π°Π½Ρ‹.<c>false</c> ΠΎΡ‚ΠΌΠ΅Π½Π° Π½Π΅ Π²ΠΎΠ·ΠΌΠΎΠΆΠ½Π°, ΠΏΠΎΡΠΊΠΎΠ»ΡŒΠΊΡƒ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ ΡƒΠΆΠ΅ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΎ ΠΈ ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊΠΈ ΠΎΡ‚Ρ€Π°Π±ΠΎΡ‚Π°Π»ΠΈ.</returns>
125 130 public bool Cancel() {
126 131 return Cancel(true);
127 132 }
128 133
129 134 /// <summary>
130 135 /// Adds new handlers to this promise.
131 136 /// </summary>
132 137 /// <param name="success">The handler of the successfully completed operation.
133 138 /// This handler will recieve an operation result as a parameter.</param>
134 139 /// <param name="error">Handles an exception that may occur during the operation.</param>
135 140 /// <returns>The new promise chained to this one.</returns>
136 141 public Promise<T> Then(ResultHandler<T> success, ErrorHandler error) {
137 142 if (success == null && error == null)
138 143 return this;
139 144
140 145 var medium = new Promise<T>(this, true);
141 146
142 147 var handlerInfo = new ResultHandlerInfo();
143 148
144 149 if (success != null)
145 150 handlerInfo.resultHandler = x => {
146 151 success(x);
147 152 medium.Resolve(x);
148 153 };
149 154 else
150 155 handlerInfo.resultHandler = medium.Resolve;
151 156
152 157 if (error != null)
153 158 handlerInfo.errorHandler = x => {
154 159 try {
155 160 error(x);
156 161 } catch { }
157 162 medium.Reject(x);
158 163 };
159 164 else
160 165 handlerInfo.errorHandler = medium.Reject;
161 166
162 167 AddHandler(handlerInfo);
163 168
164 169 return medium;
165 170 }
166 171
167 172 /// <summary>
168 173 /// Adds new handlers to this promise.
169 174 /// </summary>
170 175 /// <param name="success">The handler of the successfully completed operation.
171 176 /// This handler will recieve an operation result as a parameter.</param>
172 177 /// <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 178 /// <returns>The new promise chained to this one.</returns>
174 179 public Promise<T> Then(ResultHandler<T> success, ErrorHandler<T> error) {
175 180 if (success == null && error == null)
176 181 return this;
177 182
178 183 var medium = new Promise<T>(this, true);
179 184
180 185 var handlerInfo = new ResultHandlerInfo();
181 186
182 187 if (success != null)
183 188 handlerInfo.resultHandler = x => {
184 189 success(x);
185 190 medium.Resolve(x);
186 191 };
187 192 else
188 193 handlerInfo.resultHandler = medium.Resolve;
189 194
190 195 if (error != null)
191 196 handlerInfo.errorHandler = x => {
192 197 try {
193 198 medium.Resolve(error(x));
194 199 } catch { }
195 200 medium.Reject(x);
196 201 };
197 202 else
198 203 handlerInfo.errorHandler = medium.Reject;
199 204
200 205 AddHandler(handlerInfo);
201 206
202 207 return medium;
203 208 }
204 209
205 210
206 211 public Promise<T> Then(ResultHandler<T> success) {
207 212 if (success == null)
208 213 return this;
209 214
210 215 var medium = new Promise<T>(this, true);
211 216
212 217 var handlerInfo = new ResultHandlerInfo();
213 218
214 219 if (success != null)
215 220 handlerInfo.resultHandler = x => {
216 221 success(x);
217 222 medium.Resolve(x);
218 223 };
219 224 else
220 225 handlerInfo.resultHandler = medium.Resolve;
221 226
222 227 handlerInfo.errorHandler = medium.Reject;
223 228
224 229 AddHandler(handlerInfo);
225 230
226 231 return medium;
227 232 }
228 233
229 234 public Promise<T> Error(ErrorHandler error) {
230 235 return Then(null, error);
231 236 }
232 237
233 238 /// <summary>
234 239 /// Handles error and allows to keep the promise.
235 240 /// </summary>
236 241 /// <remarks>
237 242 /// If the specified handler throws an exception, this exception will be used to reject the promise.
238 243 /// </remarks>
239 244 /// <param name="handler">The error handler which returns the result of the promise.</param>
240 245 /// <returns>New promise.</returns>
241 246 public Promise<T> Error(ErrorHandler<T> handler) {
242 247 if (handler == null)
243 248 return this;
244 249
245 250 var medium = new Promise<T>(this, true);
246 251
247 252 AddHandler(new ResultHandlerInfo {
248 253 errorHandler = e => {
249 254 try {
250 255 medium.Resolve(handler(e));
251 256 } catch (Exception e2) {
252 257 medium.Reject(e2);
253 258 }
254 259 }
255 260 });
256 261
257 262 return medium;
258 263 }
259 264
260 265 public Promise<T> Anyway(Action handler) {
261 266 if (handler == null)
262 267 return this;
263 268
264 269 var medium = new Promise<T>();
265 270
266 271 AddHandler(new ResultHandlerInfo {
267 272 resultHandler = x => {
268 273 // to avoid handler being called multiple times we handle exception by ourselfs
269 274 try {
270 275 handler();
271 276 medium.Resolve(x);
272 277 } catch (Exception e) {
273 278 medium.Reject(e);
274 279 }
275 280 },
276 281 errorHandler = x => {
277 282 try {
278 283 handler();
279 284 } catch { }
280 285 medium.Reject(x);
281 286 }
282 287 });
283 288
284 289 return medium;
285 290 }
286 291
287 292 /// <summary>
288 293 /// ΠŸΠΎΠ·Π²ΠΎΠ»ΡΠ΅Ρ‚ ΠΏΡ€Π΅ΠΎΠ±Ρ€Π°Π·ΠΎΠ²Π°Ρ‚ΡŒ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚ выполСния ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ ΠΊ Π½ΠΎΠ²ΠΎΠΌΡƒ Ρ‚ΠΈΠΏΡƒ.
289 294 /// </summary>
290 295 /// <typeparam name="TNew">Новый Ρ‚ΠΈΠΏ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Π°.</typeparam>
291 296 /// <param name="mapper">ΠŸΡ€Π΅ΠΎΠ±Ρ€Π°Π·ΠΎΠ²Π°Π½ΠΈΠ΅ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Π° ΠΊ Π½ΠΎΠ²ΠΎΠΌΡƒ Ρ‚ΠΈΠΏΡƒ.</param>
292 297 /// <param name="error">ΠžΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊ ошибки. Π”Π°Π½Π½Ρ‹ΠΉ ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊ ΠΏΠΎΠ»ΡƒΡ‡ΠΈΡ‚
293 298 /// ΠΈΡΠΊΠ»ΡŽΡ‡Π΅Π½ΠΈΠ΅ возникшСС ΠΏΡ€ΠΈ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΈΠΈ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ.</param>
294 299 /// <returns>НовоС ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅, ΠΊΠΎΡ‚ΠΎΡ€ΠΎΠ΅ Π±ΡƒΠ΄Π΅Ρ‚ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΎ ΠΏΡ€ΠΈ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΈΠΈ исходного обСщания.</returns>
295 300 public Promise<TNew> Map<TNew>(ResultMapper<T, TNew> mapper, ErrorHandler error) {
296 301 if (mapper == null)
297 302 throw new ArgumentNullException("mapper");
298 303
299 304 // создаСм ΠΏΡ€ΠΈΡ†Π΅ΠΏΠ»Π΅Π½Π½ΠΎΠ΅ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅
300 305 var chained = new Promise<TNew>();
301 306
302 307 AddHandler(new ResultHandlerInfo() {
303 308 resultHandler = result => chained.Resolve(mapper(result)),
304 309 errorHandler = delegate(Exception e) {
305 310 if (error != null)
306 311 try {
307 312 error(e);
308 313 } catch { }
309 314 // Π² случаС ошибки Π½ΡƒΠΆΠ½ΠΎ ΠΏΠ΅Ρ€Π΅Π΄Π°Ρ‚ΡŒ ΠΈΡΠΊΠ»ΡŽΡ‡Π΅Π½ΠΈΠ΅ дальшС ΠΏΠΎ Ρ†Π΅ΠΏΠΎΡ‡ΠΊΠ΅
310 315 chained.Reject(e);
311 316 }
312 317 });
313 318
314 319 return chained;
315 320 }
316 321
317 322 public Promise<TNew> Map<TNew>(ResultMapper<T, TNew> mapper) {
318 323 return Map(mapper, null);
319 324 }
320 325
321 326 /// <summary>
322 327 /// БцСпляСт нСсколько аснхронных ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΉ. Указанная асинхронная опСрация Π±ΡƒΠ΄Π΅Ρ‚ Π²Ρ‹Π·Π²Π°Π½Π° послС
323 328 /// выполнСния Ρ‚Π΅ΠΊΡƒΡ‰Π΅ΠΉ, Π° Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚ Ρ‚Π΅ΠΊΡƒΡ‰Π΅ΠΉ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ ΠΌΠΎΠΆΠ΅Ρ‚ Π±Ρ‹Ρ‚ΡŒ использован для ΠΈΠ½ΠΈΡ†ΠΈΠ°Π»ΠΈΠ·Π°Ρ†ΠΈΠΈ
324 329 /// Π½ΠΎΠ²ΠΎΠΉ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ.
325 330 /// </summary>
326 331 /// <typeparam name="TNew">Π’ΠΈΠΏ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Π° ΡƒΠΊΠ°Π·Π°Π½Π½ΠΎΠΉ асинхронной ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ.</typeparam>
327 332 /// <param name="chained">Асинхронная опСрация, которая Π΄ΠΎΠ»ΠΆΠ½Π° Π±ΡƒΠ΄Π΅Ρ‚ Π½Π°Ρ‡Π°Ρ‚ΡŒΡΡ послС выполнСния Ρ‚Π΅ΠΊΡƒΡ‰Π΅ΠΉ.</param>
328 333 /// <param name="error">ΠžΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊ ошибки. Π”Π°Π½Π½Ρ‹ΠΉ ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊ ΠΏΠΎΠ»ΡƒΡ‡ΠΈΡ‚
329 334 /// ΠΈΡΠΊΠ»ΡŽΡ‡Π΅Π½ΠΈΠ΅ возникшСС ΠΏΡ€ΠΈ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΈΠΈ Ρ‚Π΅ΠΊΡƒΠ΅Ρ‰ΠΉ ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ.</param>
330 335 /// <returns>НовоС ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅, ΠΊΠΎΡ‚ΠΎΡ€ΠΎΠ΅ Π±ΡƒΠ΄Π΅Ρ‚ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΎ ΠΏΠΎ ΠΎΠΊΠΎΠ½Ρ‡Π°Π½ΠΈΡŽ ΡƒΠΊΠ°Π·Π°Π½Π½ΠΎΠΉ аснхронной ΠΎΠΏΠ΅Ρ€Π°Ρ†ΠΈΠΈ.</returns>
331 336 public Promise<TNew> Chain<TNew>(ChainedOperation<T, TNew> chained, ErrorHandler error) {
332 337
333 338 // ΠΏΡ€ΠΎΠ±Π»Π΅ΠΌΠ° Π² Ρ‚ΠΎΠΌ, Ρ‡Ρ‚ΠΎ Π½Π° ΠΌΠΎΠΌΠ΅Π½Ρ‚ связывания Π΅Ρ‰Π΅ Π½Π΅ Π½Π°Ρ‡Π°Ρ‚Π° асинхронная опСрация, поэтому Π½ΡƒΠΆΠ½ΠΎ
334 339 // ΡΠΎΠ·Π΄Π°Ρ‚ΡŒ посрСдника, ΠΊ ΠΊΠΎΡ‚ΠΎΡ€ΠΎΠΌΡƒ Π±ΡƒΠ΄ΡƒΡ‚ ΠΏΠΎΠ΄Π²Ρ‹Π·ΡΠ²Π°Ρ‚ΡŒΡΡ ΡΠ»Π΅Π΄ΡƒΡŽΡ‰ΠΈΠ΅ ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊΠΈ.
335 340 // ΠΊΠΎΠ³Π΄Π° Π±ΡƒΠ΄Π΅Ρ‚ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½Π° Ρ€Π΅Π°Π»ΡŒΠ½Π°Ρ асинхронная опСрация, ΠΎΠ½Π° ΠΎΠ±Ρ€Π°Ρ‚ΠΈΡ‚ΡŒΡΡ ΠΊ посрСднику, Ρ‡Ρ‚ΠΎΠ±Ρ‹
336 341 // ΠΏΠ΅Ρ€Π΅Π΄Π°Ρ‚ΡŒ Ρ‡Π΅Ρ€Π΅Π· Π½Π΅Π³ΠΎ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Ρ‹ Ρ€Π°Π±ΠΎΡ‚Ρ‹.
337 342 var medium = new Promise<TNew>(this, true);
338 343
339 344 AddHandler(new ResultHandlerInfo {
340 345 resultHandler = delegate(T result) {
341 346 if (medium.State == PromiseState.Cancelled)
342 347 return;
343 348
344 349 var promise = chained(result);
345 350
346 351 // notify chained operation that it's not needed
347 352 medium.Cancelled(() => promise.Cancel());
348 353 promise.Then(
349 354 x => medium.Resolve(x),
350 355 e => medium.Reject(e)
351 356 );
352 357 },
353 358 errorHandler = delegate(Exception e) {
354 359 if (error != null)
355 360 error(e);
356 361 // Π² случаС ошибки Π½ΡƒΠΆΠ½ΠΎ ΠΏΠ΅Ρ€Π΅Π΄Π°Ρ‚ΡŒ ΠΈΡΠΊΠ»ΡŽΡ‡Π΅Π½ΠΈΠ΅ дальшС ΠΏΠΎ Ρ†Π΅ΠΏΠΎΡ‡ΠΊΠ΅
357 362 medium.Reject(e);
358 363 }
359 364 });
360 365
361 366 return medium;
362 367 }
363 368
364 369 public Promise<TNew> Chain<TNew>(ChainedOperation<T, TNew> chained) {
365 370 return Chain(chained, null);
366 371 }
367 372
368 373 public Promise<T> Cancelled(Action handler) {
369 374 if (handler == null)
370 375 return this;
371 376 lock (m_lock) {
372 377 if (m_state == PromiseState.Unresolved)
373 378 m_cancelHandlers.AddLast(handler);
374 379 else if (m_state == PromiseState.Cancelled)
375 380 handler();
376 381 }
377 382 return this;
378 383 }
379 384
380 385 public void HandleCancelled(Action handler) {
381 386 Cancelled(handler);
382 387 }
383 388
384 389 /// <summary>
385 390 /// ДоТидаСтся ΠΎΡ‚Π»ΠΎΠΆΠ΅Π½Π½ΠΎΠ³ΠΎ обСщания ΠΈ Π² случаС успСха, Π²ΠΎΠ·Π²Ρ€Π°Ρ‰Π°Π΅Ρ‚
386 391 /// Π΅Π³ΠΎ, Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚, Π² ΠΏΡ€ΠΎΡ‚ΠΈΠ²Π½ΠΎΠΌ случаС бросаСт ΠΈΡΠΊΠ»ΡŽΡ‡Π΅Π½ΠΈΠ΅.
387 392 /// </summary>
388 393 /// <remarks>
389 394 /// <para>
390 395 /// Если ΠΎΠΆΠΈΠ΄Π°Π½ΠΈΠ΅ обСщания Π±Ρ‹Π»ΠΎ ΠΏΡ€Π΅Ρ€Π²Π°Π½ΠΎ ΠΏΠΎ Ρ‚Π°ΠΉΠΌΠ°ΡƒΡ‚Ρƒ, это Π½Π΅ Π·Π½Π°Ρ‡ΠΈΡ‚,
391 396 /// Ρ‡Ρ‚ΠΎ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ Π±Ρ‹Π»ΠΎ ΠΎΡ‚ΠΌΠ΅Π½Π΅Π½ΠΎ ΠΈΠ»ΠΈ Ρ‡Ρ‚ΠΎ-Ρ‚ΠΎ Π² этом Ρ€ΠΎΠ΄Π΅, это Ρ‚ΠΎΠ»ΡŒΠΊΠΎ
392 397 /// ΠΎΠ·Π½Π°Ρ‡Π°Π΅Ρ‚, Ρ‡Ρ‚ΠΎ ΠΌΡ‹ Π΅Π³ΠΎ Π½Π΅ доТдались, ΠΎΠ΄Π½Π°ΠΊΠΎ всС зарСгистрированныС
393 398 /// ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊΠΈ, ΠΊΠ°ΠΊ Π±Ρ‹Π»ΠΈ Ρ‚Π°ΠΊ ΠΎΡΡ‚Π°Π»ΠΈΡΡŒ ΠΈ ΠΎΠ½ΠΈ Π±ΡƒΠ΄ΡƒΡ‚ Π²Ρ‹Π·Π²Π°Π½Ρ‹, ΠΊΠΎΠ³Π΄Π°
394 399 /// ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ Π±ΡƒΠ΄Π΅Ρ‚ Π²Ρ‹ΠΏΠΎΠ»Π½Π΅Π½ΠΎ.
395 400 /// </para>
396 401 /// <para>
397 402 /// Π’Π°ΠΊΠΎΠ΅ ΠΏΠΎΠ²Π΅Π΄Π΅Π½ΠΈΠ΅ Π²ΠΏΠΎΠ»Π½Π΅ ΠΎΠΏΡ€Π°Π²Π΄Π°Π½ΠΎ ΠΏΠΎΡΠΊΠΎΠ»ΡŒΠΊΡƒ Ρ‚Π°ΠΉΠΌΠ°ΡƒΡ‚ ΠΌΠΎΠΆΠ΅Ρ‚ ΠΈΡΡ‚Π΅Ρ‡ΡŒ
398 403 /// Π² Ρ‚ΠΎΡ‚ ΠΌΠΎΠΌΠ΅Π½Ρ‚, ΠΊΠΎΠ³Π΄Π° Π½Π°Ρ‡Π°Π»Π°ΡΡŒ ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚ΠΊΠ° Ρ†Π΅ΠΏΠΎΡ‡ΠΊΠΈ ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊΠΎΠ², ΠΈ
399 404 /// ΠΊ Ρ‚ΠΎΠΌΡƒ ΠΆΠ΅ Ρ‚Π΅ΠΊΡƒΡ‰Π΅Π΅ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠ΅ ΠΌΠΎΠΆΠ΅Ρ‚ ΡΡ‚ΠΎΡΡ‚ΡŒ Π² Ρ†Π΅ΠΏΠΎΡ‡ΠΊΠ΅ ΠΎΠ±Π΅Ρ‰Π°Π½ΠΈΠΉ ΠΈ Π΅Π³ΠΎ
400 405 /// ΠΎΡ‚ΠΊΠ»ΠΎΠ½Π΅Π½ΠΈΠ΅ ΠΌΠΎΠΆΠ΅Ρ‚ привСсти ΠΊ Π½Π΅ΠΏΡ€ΠΎΠ³Π½ΠΎΠ·ΠΈΡ€ΡƒΠ΅ΠΌΠΎΠΌΡƒ Ρ€Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚Ρƒ.
401 406 /// </para>
402 407 /// </remarks>
403 408 /// <param name="timeout">ВрСмя оТидания</param>
404 409 /// <returns>Π Π΅Π·ΡƒΠ»ΡŒΡ‚Π°Ρ‚ выполнСния обСщания</returns>
405 410 public T Join(int timeout) {
406 411 var evt = new ManualResetEvent(false);
407 412 Anyway(() => evt.Set());
408 413 Cancelled(() => evt.Set());
409 414
410 415 if (!evt.WaitOne(timeout, true))
411 416 throw new TimeoutException();
412 417
413 418 switch (State) {
414 419 case PromiseState.Resolved:
415 420 return m_result;
416 421 case PromiseState.Cancelled:
417 422 throw new OperationCanceledException();
418 423 case PromiseState.Rejected:
419 424 throw new TargetInvocationException(m_error);
420 425 default:
421 426 throw new ApplicationException(String.Format("Invalid promise state {0}", State));
422 427 }
423 428 }
424 429
425 430 public T Join() {
426 431 return Join(Timeout.Infinite);
427 432 }
428 433
429 434 void AddHandler(ResultHandlerInfo handler) {
430 435 bool invokeRequired = false;
431 436
432 437 lock (m_lock) {
433 438 m_childrenCount++;
434 439 if (m_state == PromiseState.Unresolved) {
435 440 m_resultHandlers.AddLast(handler);
436 441 } else
437 442 invokeRequired = true;
438 443 }
439 444
440 445 // ΠΎΠ±Ρ€Π°Π±ΠΎΡ‚Ρ‡ΠΈΠΊΠΈ Π½Π΅ Π΄ΠΎΠ»ΠΆΠ½Ρ‹ Π±Π»ΠΎΠΊΠΈΡ€ΠΎΠ²Π°Ρ‚ΡŒ сам ΠΎΠ±ΡŠΠ΅ΠΊΡ‚
441 446 if (invokeRequired)
442 447 InvokeHandler(handler);
443 448 }
444 449
445 450 void InvokeHandler(ResultHandlerInfo handler) {
446 451 switch (m_state) {
447 452 case PromiseState.Resolved:
448 453 try {
449 454 if (handler.resultHandler != null)
450 455 handler.resultHandler(m_result);
451 456 } catch (Exception e) {
452 457 try {
453 458 if (handler.errorHandler != null)
454 459 handler.errorHandler(e);
455 460 } catch { }
456 461 }
457 462 break;
458 463 case PromiseState.Rejected:
459 464 try {
460 465 if (handler.errorHandler != null)
461 466 handler.errorHandler(m_error);
462 467 } catch { }
463 468 break;
464 469 default:
465 470 // do nothing
466 471 return;
467 472 }
468 473 }
469 474
470 475 protected virtual void OnStateChanged() {
471 476 switch (m_state) {
472 477 case PromiseState.Resolved:
473 478 foreach (var resultHandlerInfo in m_resultHandlers)
474 479 try {
475 480 if (resultHandlerInfo.resultHandler != null)
476 481 resultHandlerInfo.resultHandler(m_result);
477 482 } catch (Exception e) {
478 483 try {
479 484 if (resultHandlerInfo.errorHandler != null)
480 485 resultHandlerInfo.errorHandler(e);
481 486 } catch { }
482 487 }
483 488 break;
484 489 case PromiseState.Cancelled:
485 490 foreach (var cancelHandler in m_cancelHandlers)
486 491 cancelHandler();
487 492 break;
488 493 case PromiseState.Rejected:
489 494 foreach (var resultHandlerInfo in m_resultHandlers)
490 495 try {
491 496 if (resultHandlerInfo.errorHandler != null)
492 497 resultHandlerInfo.errorHandler(m_error);
493 498 } catch { }
494 499 break;
495 500 default:
496 501 throw new InvalidOperationException(String.Format("Promise entered an invalid state {0}", m_state));
497 502 }
498 503
499 504 m_resultHandlers = null;
500 505 m_cancelHandlers = null;
501 506 }
502 507
503 508
504 509
505 510 public bool IsExclusive {
506 511 get {
507 512 lock (m_lock) {
508 513 return m_childrenCount <= 1;
509 514 }
510 515 }
511 516 }
512 517
513 518 public PromiseState State {
514 519 get {
515 520 lock (m_lock) {
516 521 return m_state;
517 522 }
518 523 }
519 524 }
520 525
521 526 protected bool Cancel(bool dependencies) {
522 527 bool result;
523 528
524 529 lock (m_lock) {
525 530 if (m_state == PromiseState.Unresolved) {
526 531 m_state = PromiseState.Cancelled;
527 532 result = true;
528 533 } else {
529 534 result = false;
530 535 }
531 536 }
532 537
533 538 if (result)
534 539 OnStateChanged();
535 540
536 541 if (dependencies && m_parent != null && m_parent.IsExclusive) {
537 542 m_parent.Cancel();
538 543 }
539 544
540 545 return result;
541 546 }
542 547
543 548 }
544 549 }
General Comments 0
You need to be logged in to leave comments. Login now