implab/ImplabNet Commit - r149:eb793fbbe4ea

fixed promises cancellation

cin -

r149:eb793fbbe4ea v2

parent child

Context file:

r149:eb793fbbe4ea

Collapse all files

Implab.Test/AsyncTests.cs +8 -10

             using System;
             using System.Reflection;
             using System.Threading;
             using Implab.Parallels;
             #if MONO
             using NUnit.Framework;
             using TestClassAttribute = NUnit.Framework.TestFixtureAttribute;
             using TestMethodAttribute = NUnit.Framework.TestAttribute;
             #else
             using Microsoft.VisualStudio.TestTools.UnitTesting;
             #endif
             namespace Implab.Test {
                 [TestClass]
                 public class AsyncTests {
                     [TestMethod]
                     public void ResolveTest() {
                         int res = -1;
                         var p = new Promise<int>();
                         p.Then(x => res = x);
                         p.Resolve(100);
                         Assert.AreEqual(100, res);
                     }
                     [TestMethod]
                     public void RejectTest() {
                         int res = -1;
                         Exception err = null;
                         var p = new Promise<int>();
                         p.Then(
                             x => res = x,
                             e => {
                                 err = e;
                                 return -2;
                             }
                         );
                         p.Reject(new ApplicationException("error"));
                         Assert.AreEqual(res, -1);
                         Assert.AreEqual(err.Message, "error");
                     }
                     [TestMethod]
                     public void CancelExceptionTest() {
                         var p = new Promise<bool>();
                         p.CancelOperation(null);
                         var p2 = p.Then(x => x, null, reason => {
                             throw new ApplicationException("CANCELLED");
                         });
                         try {
                             p2.Join();
                             Assert.Fail();
                         } catch (ApplicationException err) {
                             Assert.AreEqual("CANCELLED", err.InnerException.Message);
                         }
                     }
                     [TestMethod]
                     public void ContinueOnCancelTest() {
                         var p = new Promise<bool>();
                         p.CancelOperation(null);
                         var p2 = p
                             .Then(x => x, null, reason => {
                                 throw new ApplicationException("CANCELLED");
                             })
                             .Then(x => x, e => true);
                         Assert.AreEqual(true, p2.Join());
                     }
                     [TestMethod]
                     public void JoinSuccessTest() {
                         var p = new Promise<int>();
                         p.Resolve(100);
                         Assert.AreEqual(p.Join(), 100);
                     }
                     [TestMethod]
                     public void JoinFailTest() {
                         var p = new Promise<int>();
                         p.Reject(new ApplicationException("failed"));
                         try {
                             p.Join();
                             throw new ApplicationException("WRONG!");
                         } catch (TargetInvocationException err) {
                             Assert.AreEqual(err.InnerException.Message, "failed");
                         } catch {
                             Assert.Fail("Got wrong excaption");
                         }
                     }
                     [TestMethod]
                     public void MapTest() {
                         var p = new Promise<int>();
                         var p2 = p.Then(x => x.ToString());
                         p.Resolve(100);
                         Assert.AreEqual(p2.Join(), "100");
                     }
                     [TestMethod]
                     public void FixErrorTest() {
                         var p = new Promise<int>();
                         var p2 = p.Then(x => x, e => 101);
                         p.Reject(new Exception());
                         Assert.AreEqual(p2.Join(), 101);
                     }
                     [TestMethod]
                     public void ChainTest() {
                         var p1 = new Promise<int>();
                         var p3 = p1.Chain(x => {
                             var p2 = new Promise<string>();
                             p2.Resolve(x.ToString());
                             return p2;
                         });
                         p1.Resolve(100);
                         Assert.AreEqual(p3.Join(), "100");
                     }
                     [TestMethod]
                     public void ChainFailTest() {
                         var p1 = new Promise<int>();
                         var p3 = p1.Chain(x => {
                             var p2 = new Promise<string>();
                             p2.Reject(new Exception("DIE!!!"));
                             return p2;
                         });
                         p1.Resolve(100);
                         Assert.IsTrue(p3.IsResolved);
                     }
                     [TestMethod]
                     public void PoolTest() {
                         var pid = Thread.CurrentThread.ManagedThreadId;
                         var p = AsyncPool.Invoke(() => Thread.CurrentThread.ManagedThreadId);
                         Assert.AreNotEqual(pid, p.Join());
                     }
                     [TestMethod]
                     public void WorkerPoolSizeTest() {
                         var pool = new WorkerPool(5, 10, 1);
                         Assert.AreEqual(5, pool.PoolSize);
                         pool.Invoke(() => { Thread.Sleep(100000000); return 10; });
                         pool.Invoke(() => { Thread.Sleep(100000000); return 10; });
                         pool.Invoke(() => { Thread.Sleep(100000000); return 10; });
                         Assert.AreEqual(5, pool.PoolSize);
                         for (int i = 0; i < 100; i++)
                             pool.Invoke(() => { Thread.Sleep(100000000); return 10; });
                         Thread.Sleep(200);
                         Assert.AreEqual(10, pool.PoolSize);
                         pool.Dispose();
                     }
                     [TestMethod]
                     public void WorkerPoolCorrectTest() {
                         var pool = new WorkerPool(0,1000,100);
                         const int iterations = 1000;
                         int pending = iterations;
                         var stop = new ManualResetEvent(false);
                         var count = 0;
                         for (int i = 0; i < iterations; i++) {
                             pool
                                 .Invoke(() => 1)
                                 .Then(x => Interlocked.Add(ref count, x))
                                 .Then(x => Math.Log10(x))
                                 .On(() => {
                                     Interlocked.Decrement(ref pending);
                                     if (pending == 0)
                                         stop.Set();
                                 }, PromiseEventType.All);
                         }
                         stop.WaitOne();
                         Assert.AreEqual(iterations, count);
                         Console.WriteLine("Max threads: {0}", pool.MaxRunningThreads);
                         pool.Dispose();
                     }
                     [TestMethod]
                     public void WorkerPoolDisposeTest() {
                         var pool = new WorkerPool(5, 20);
                         Assert.AreEqual(5, pool.PoolSize);
                         pool.Dispose();
                         Thread.Sleep(500);
                         Assert.AreEqual(0, pool.PoolSize);
                         pool.Dispose();
                     }
                     [TestMethod]
                     public void MTQueueTest() {
                         var queue = new MTQueue<int>();
                         int res;
                         queue.Enqueue(10);
                         Assert.IsTrue(queue.TryDequeue(out res));
                         Assert.AreEqual(10, res);
                         Assert.IsFalse(queue.TryDequeue(out res));
                         for (int i = 0; i < 1000; i++)
                             queue.Enqueue(i);
                         for (int i = 0; i < 1000; i++) {
                             queue.TryDequeue(out res);
                             Assert.AreEqual(i, res);
                         }
                         int writers = 0;
                         int readers = 0;
                         var stop = new ManualResetEvent(false);
                         int total = 0;
                         const int itemsPerWriter = 10000;
                         const int writersCount = 10;
                         for (int i = 0; i < writersCount; i++) {
                             Interlocked.Increment(ref writers);
                             AsyncPool
                                 .RunThread(() => {
                                     for (int ii = 0; ii < itemsPerWriter; ii++) {
                                         queue.Enqueue(1);
                                     }
                                     return 1;
                                 })
                                 .On(() => Interlocked.Decrement(ref writers), PromiseEventType.All);
                         }
                         for (int i = 0; i < 10; i++) {
                             Interlocked.Increment(ref readers);
                             AsyncPool
                                 .RunThread(() => {
                                     int t;
                                     do {
                                         while (queue.TryDequeue(out t))
                                             Interlocked.Add(ref total, t);
                                     } while (writers > 0);
                                     return 1;
                                 })
                                 .On(() => {
                                     Interlocked.Decrement(ref readers);
                                     if (readers == 0)
                                         stop.Set();
                                 }, PromiseEventType.All);
                         }
                         stop.WaitOne();
                         Assert.AreEqual(100000, total);
                     }
                     [TestMethod]
                     public void AsyncQueueTest() {
                         var queue = new AsyncQueue<int>();
                         int res;
                         queue.Enqueue(10);
                         Assert.IsTrue(queue.TryDequeue(out res));
                         Assert.AreEqual(10, res);
                         Assert.IsFalse(queue.TryDequeue(out res));
                         for (int i = 0; i < 1000; i++)
                             queue.Enqueue(i);
                         for (int i = 0; i < 1000; i++) {
                             queue.TryDequeue(out res);
                             Assert.AreEqual(i, res);
                         }
                         const int count = 10000000;
                         int res1 = 0, res2 = 0;
                         var t1 = Environment.TickCount;
                         AsyncPool.RunThread(
                             () => {
                                 for (var i = 0; i < count; i++)
                                     queue.Enqueue(1);
                                 Console.WriteLine("done writer #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 for (var i = 0; i < count; i++)
                                     queue.Enqueue(2);
                                 Console.WriteLine("done writer #2: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 int temp;
                                 int i = 0;
                                 while (i < count)
                                     if (queue.TryDequeue(out temp)) {
                                         i++;
                                         res1 += temp;
                                     }
                                 Console.WriteLine("done reader #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 int temp;
                                 int i = 0;
                                 while (i < count)
                                     if (queue.TryDequeue(out temp)) {
                                         i++;
                                         res2 += temp;
                                     }
                                 Console.WriteLine("done reader #2: {0} ms", Environment.TickCount - t1);
                             }
                         )
                             .Bundle()
                             .Join();
                         Assert.AreEqual(count * 3, res1 + res2);
                         Console.WriteLine(
                             "done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}",
                             Environment.TickCount - t1,
                             res1,
                             res2,
                             res1 + res2,
                             count
                         );
                     }
                     [TestMethod]
                     public void AsyncQueueBatchTest() {
                         var queue = new AsyncQueue<int>();
                         const int wBatch = 29;
                         const int wCount = 400000;
                         const int total = wBatch * wCount * 2;
                         const int summ = wBatch * wCount * 3;
                         int r1 = 0, r2 = 0;
                         const int rBatch = 111;
                         int read = 0;
                         var t1 = Environment.TickCount;
                         AsyncPool.RunThread(
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 1;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 2;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #2: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[rBatch];
                                 while(read < total) {
                                     int actual;
                                     if (queue.TryDequeueRange(buffer,0,rBatch,out actual)) {
                                         for(int i=0; i< actual; i++)
                                             r1 += buffer[i];
                                         Interlocked.Add(ref read, actual);
                                     }
                                 }
                                 Console.WriteLine("done reader #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[rBatch];
                                 while(read < total) {
                                     int actual;
                                     if (queue.TryDequeueRange(buffer,0,rBatch,out actual)) {
                                         for(int i=0; i< actual; i++)
                                             r2 += buffer[i];
                                         Interlocked.Add(ref read, actual);
                                     }
                                 }
                                 Console.WriteLine("done reader #2: {0} ms", Environment.TickCount - t1);
                             }
                         )
                             .Bundle()
                             .Join();
                         Assert.AreEqual(summ , r1 + r2);
                         Console.WriteLine(
                             "done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}",
                             Environment.TickCount - t1,
                             r1,
                             r2,
                             r1 + r2,
                             total
                         );
                     }
                     [TestMethod]
                     public void AsyncQueueChunkDequeueTest() {
                         var queue = new AsyncQueue<int>();
                         const int wBatch = 31;
                         const int wCount = 200000;
                         const int total = wBatch * wCount * 3;
                         const int summ = wBatch * wCount * 6;
                         int r1 = 0, r2 = 0;
                         const int rBatch = 1024;
                         int read = 0;
                         var t1 = Environment.TickCount;
                         AsyncPool.RunThread(
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 1;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 2;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #2: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 3;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #3: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[rBatch];
                                 int count = 1;
                                 double avgchunk = 0;
                                 while(read < total) {
                                     int actual;
                                     if (queue.TryDequeueChunk(buffer,0,rBatch,out actual)) {
                                         for(int i=0; i< actual; i++)
                                             r2 += buffer[i];
                                         Interlocked.Add(ref read, actual);
                                         avgchunk = avgchunk*(count-1)/count + actual/(double)count;
                                         count ++;
                                     }
                                 }
                                 Console.WriteLine("done reader #2: {0} ms, avg chunk size: {1}", Environment.TickCount - t1, avgchunk);
                             }
                         )
                             .Bundle()
                             .Join();
                         Assert.AreEqual(summ , r1 + r2);
                         Console.WriteLine(
                             "done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}",
                             Environment.TickCount - t1,
                             r1,
                             r2,
                             r1 + r2,
                             total
                         );
                     }
                     [TestMethod]
                     public void AsyncQueueDrainTest() {
                         var queue = new AsyncQueue<int>();
                         const int wBatch = 11;
                         const int wCount = 200000;
                         const int total = wBatch * wCount * 3;
                         const int summ = wBatch * wCount * 3;
                         int r1 = 0, r2 = 0;
                         const int rBatch = 11;
                         int read = 0;
                         var t1 = Environment.TickCount;
                         AsyncPool.RunThread(
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 1;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 for(int i =0; i < wCount * wBatch; i++)
                                     queue.Enqueue(1);
                                 Console.WriteLine("done writer #2: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 var buffer = new int[wBatch];
                                 for(int i = 0; i<wBatch; i++)
                                     buffer[i] = 1;
                                 for(int i =0; i < wCount; i++)
                                     queue.EnqueueRange(buffer,0,wBatch);
                                 Console.WriteLine("done writer #3: {0} ms", Environment.TickCount - t1);
                             },
                             /*() => {
                                 int temp;
                                 int count = 0;
                                 while (read < total)
                                     if (queue.TryDequeue(out temp)) {
                                         count++;
                                         r1 += temp;
                                         Interlocked.Increment(ref read);
                                     }
                                 Console.WriteLine("done reader #1: {0} ms, {1} count", Environment.TickCount - t1, count);
                             },*/
                             /*() => {
                                 var buffer = new int[rBatch];
                                 var count = 0;
                                 while(read < total) {
                                     int actual;
                                     if (queue.TryDequeueRange(buffer,0,rBatch,out actual)) {
                                         for(int i=0; i< actual; i++)
                                             r1 += buffer[i];
                                         Interlocked.Add(ref read, actual);
                                         count += actual;
                                     }
                                 }
                                 Console.WriteLine("done reader #1: {0} ms, {1} items", Environment.TickCount - t1, count);
                             },*/
                             () => {
                                 var count = 0;
                                 while(read < total) {
                                     var buffer = queue.Drain();
                                     for(int i=0; i< buffer.Length; i++)
                                         r1 += buffer[i];
                                         Interlocked.Add(ref read, buffer.Length);
                                     count += buffer.Length;
                                 }
                                 Console.WriteLine("done reader #1: {0} ms, {1} items", Environment.TickCount - t1, count);
                             },
                             () => {
                                 var count = 0;
                                 while(read < total) {
                                     var buffer = queue.Drain();
                                     for(int i=0; i< buffer.Length; i++)
                                         r2 += buffer[i];
                                     Interlocked.Add(ref read, buffer.Length);
                                     count += buffer.Length;
                                 }
                                 Console.WriteLine("done reader #2: {0} ms, {1} items", Environment.TickCount - t1, count);
                             }
                         )
                             .Bundle()
                             .Join();
                         Assert.AreEqual(summ , r1 + r2);
                         Console.WriteLine(
                             "done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}",
                             Environment.TickCount - t1,
                             r1,
                             r2,
                             r1 + r2,
                             total
                         );
                     }
                     [TestMethod]
                     public void ParallelMapTest() {
                         const int count = 100000;
                         var args = new double[count];
                         var rand = new Random();
                         for (int i = 0; i < count; i++)
                             args[i] = rand.NextDouble();
                         var t = Environment.TickCount;
                         var res = args.ParallelMap(x => Math.Sin(x*x), 4).Join();
                         Console.WriteLine("Map complete in {0} ms", Environment.TickCount - t);
                         t = Environment.TickCount;
                         for (int i = 0; i < count; i++)
                             Assert.AreEqual(Math.Sin(args[i] * args[i]), res[i]);
                         Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
                     }
                     [TestMethod]
                     public void ChainedMapTest() {
                         using (var pool = new WorkerPool()) {
                             const int count = 10000;
                             var args = new double[count];
                             var rand = new Random();
                             for (int i = 0; i < count; i++)
                                 args[i] = rand.NextDouble();
                             var t = Environment.TickCount;
                             var res = args
                                 .ChainedMap(
                                     // Analysis disable once AccessToDisposedClosure
                                     x => pool.Invoke(
                                         () => Math.Sin(x * x)
                                     ),
                                 )
                                 .Join();
                             Console.WriteLine("Map complete in {0} ms", Environment.TickCount - t);
                             t = Environment.TickCount;
                             for (int i = 0; i < count; i++)
                                 Assert.AreEqual(Math.Sin(args[i] * args[i]), res[i]);
                             Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
                             Console.WriteLine("Max workers: {0}", pool.MaxRunningThreads);
                         }
                     }
                     [TestMethod]
                     public void ParallelForEachTest() {
                         const int count = 100000;
                         var args = new int[count];
                         var rand = new Random();
                         for (int i = 0; i < count; i++)
                             args[i] = (int)(rand.NextDouble() * 100);
                         int result = 0;
                         var t = Environment.TickCount;
                         args.ParallelForEach(x => Interlocked.Add(ref result, x), 4).Join();
                         Console.WriteLine("Iteration complete in {0} ms, result: {1}", Environment.TickCount - t, result);
                         int result2 = 0;
                         t = Environment.TickCount;
                         for (int i = 0; i < count; i++)
                             result2 += args[i];
                         Assert.AreEqual(result2, result);
                         Console.WriteLine("Verified in {0} ms", Environment.TickCount - t);
                     }
                     [TestMethod]
                     public void ComplexCase1Test() {
                         var flags = new bool[3];
                         // op1 (aync 200ms) => op2 (async 200ms) => op3 (sync map)
                         var step1 = PromiseHelper
                             .Sleep(200, "Alan")
                             .On(() => flags[0] = true, PromiseEventType.Cancelled);
                         var p = step1
                             .Chain(x =>
                                 PromiseHelper
                                     .Sleep(200, "Hi, " + x)
                                     .Then(y => y)
                                     .On(() => flags[1] = true, PromiseEventType.Cancelled)
                             )
                             .On(() => flags[2] = true, PromiseEventType.Cancelled);
                         step1.Join();
                         p.Cancel();
                         try {
                             Assert.AreEqual(p.Join(), "Hi, Alan");
                             Assert.Fail("Shouldn't get here");
                         } catch (OperationCanceledException) {
                         }
                         Assert.IsFalse(flags[0]);
                         Assert.IsTrue(flags[1]);
                         Assert.IsTrue(flags[2]);
                     }
                     [TestMethod]
                     public void ChainedCancel1Test() {
                         // при отмене сцепленной асинхронной операции все обещание должно
                         // завершаться ошибкой OperationCanceledException
                         var p = PromiseHelper
                             .Sleep(1, "Hi, HAL!")
                             .Then(x => {
                                 // запускаем две асинхронные операции
                                 var result = PromiseHelper.Sleep(1000, "HEM ENABLED!!!");
                                 // вторая операция отменяет первую до завершения
                                 PromiseHelper
                                     .Sleep(100, "HAL, STOP!")
                                     .Then(result.Cancel);
                                 return result;
                             });
                         try {
                             p.Join();
                         } catch (TargetInvocationException err) {
                             Assert.IsTrue(err.InnerException is OperationCanceledException);
                         }
                     }
                     [TestMethod]
                     public void ChainedCancel2Test() {
                         // при отмене цепочки обещаний, вложенные операции также должны отменяться
                         var pSurvive = new Promise<bool>();
-                        var hemStarted = new ManualResetEvent(false);
+                        var hemStarted = new Signal();
                         var p = PromiseHelper
                             .Sleep(1, "Hi, HAL!")
-                            .Chain(x => {
+                            .Chain(() => {
                                 hemStarted.Set();
                                 // запускаем две асинхронные операции
                                 var result = PromiseHelper
-                                    .Sleep(100000000, "HEM ENABLED!!!")
+                                    .Sleep(2000, "HEM ENABLED!!!")
-                                    .Then(s => {
+                                    .Then(() => pSurvive.Resolve(false));
-                                        pSurvive.Resolve(false);
-                                        return s;
-                                    });
                                 result
                                     .On(() => pSurvive.Resolve(true), PromiseEventType.Cancelled);
                                 return result;
                             });
-                        hemStarted.WaitOne();
+                        hemStarted.Wait();
                         p.Cancel();
                         try {
                             p.Join();
+                            Assert.Fail();
                         } catch (OperationCanceledException) {
-                            Assert.IsTrue(pSurvive.Join());
                         }
+                        Assert.IsTrue(pSurvive.Join());
                     }
                     [TestMethod]
                     public void SharedLockTest() {
                         var l = new SharedLock();
                         int shared = 0;
                         int exclusive = 0;
                         var s1 = new Signal();
                         var log = new AsyncQueue<string>();
                         try {
                             AsyncPool.RunThread(
                                 () => {
                                     log.Enqueue("Reader #1 started");
                                     try {
                                         l.LockShared();
                                         log.Enqueue("Reader #1 lock got");
                                         if (Interlocked.Increment(ref shared) == 2)
                                             s1.Set();
                                         s1.Wait();
                                         log.Enqueue("Reader #1 finished");
                                         Interlocked.Decrement(ref shared);
                                     } finally {
                                         l.Release();
                                         log.Enqueue("Reader #1 lock released");
                                     }
                                 },
                                 () => {
                                     log.Enqueue("Reader #2 started");
                                     try {
                                         l.LockShared();
                                         log.Enqueue("Reader #2 lock got");
                                         if (Interlocked.Increment(ref shared) == 2)
                                             s1.Set();
                                         s1.Wait();
                                         log.Enqueue("Reader #2 upgrading to writer");
                                         Interlocked.Decrement(ref shared);
                                         l.Upgrade();
                                         log.Enqueue("Reader #2 upgraded");
                                         Assert.AreEqual(1, Interlocked.Increment(ref exclusive));
                                         Assert.AreEqual(0, shared);
                                         log.Enqueue("Reader #2 finished");
                                         Interlocked.Decrement(ref exclusive);
                                     } finally {
                                         l.Release();
                                         log.Enqueue("Reader #2 lock released");
                                     }
                                 },
                                 () => {
                                     log.Enqueue("Writer #1 started");
                                     try {
                                         l.LockExclusive();
                                         log.Enqueue("Writer #1 got the lock");
                                         Assert.AreEqual(1, Interlocked.Increment(ref exclusive));
                                         Interlocked.Decrement(ref exclusive);
                                         log.Enqueue("Writer #1 is finished");
                                     } finally {
                                         l.Release();
                                         log.Enqueue("Writer #1 lock released");
                                     }
                                 }
                             ).Bundle().Join(1000);
                             log.Enqueue("Done");
                         } catch(Exception error) {
                             log.Enqueue(error.Message);
                             throw;
                         } finally {
                             foreach (var m in log)
                                 Console.WriteLine(m);
                         }
                     }
                 }
             }

Implab.Test/CancelationTests.cs +6 -2

             using System;
             using Implab.Parallels;
             #if MONO
             using NUnit.Framework;
             using TestClassAttribute = NUnit.Framework.TestFixtureAttribute;
             using TestMethodAttribute = NUnit.Framework.TestAttribute;
             #else
             using Microsoft.VisualStudio.TestTools.UnitTesting;
             #endif
             namespace Implab.Test {
                 [TestClass]
                 public class CancelationTests {
                     [TestMethod]
                     public void PromiseCancelTest() {
                         var p = new Promise();
                         bool requested = false;
                         var reason = new Exception("Test");
                         // request cancelation
                         p.Cancel(reason);
                         Assert.IsTrue(p.IsCancellationRequested);
                         Assert.AreSame(reason, p.CancellationReason);
                         Assert.IsFalse(p.IsCancelled);
                         p.CancellationRequested(r => {
                             Assert.AreSame(reason, r);
                             requested = true;
                         });
                         Assert.IsTrue(requested);
                         // cancel the promise
                         Assert.IsTrue(p.CancelOperationIfRequested());
                         Assert.IsTrue(p.IsCancelled);
                         Assert.AreSame(reason, p.Error);
                     }
                     [TestMethod]
                     public void CancelActionBeforeStartTask() {
                         bool run = false;
                         var task = new ActionTask(() => {
                             run = true;
-                        }, null, null);
+                        }, null, null, true);
                         // request cancelation
                         task.Cancel();
                         Assert.IsTrue(task.IsCancelled);
                         task.Resolve();
                         Assert.IsFalse(run);
                     }
                     [TestMethod]
                     public void CancelActionAfterTaskStarted() {
                         var finish = new Signal();
                         var started = new Signal();
                         var task = new ActionTask(() => {
                             started.Set();
                             finish.Wait();
-                        }, null, null);
+                        }, null, null, true);
                         AsyncPool.RunThread(() => {
                             task.Resolve();
                         });
                         started.Wait(1000);
                         task.Cancel();
                         Assert.IsTrue(task.IsCancellationRequested);
                         Assert.IsFalse(task.IsCancelled);
                         Assert.IsFalse(task.IsResolved);
                         finish.Set();
                         task.Join(1000);
                     }
                     [TestMethod]
                     public void CancelTaskChainFromBottom() {
+                        var started = new Signal();
                         var check1 = new Signal();
                         var requested = false;
                         var p1 = AsyncPool.RunThread(token => {
                             token.CancellationRequested(reason => requested = true);
+                            started.Set();
                             check1.Wait();
                             token.CancelOperationIfRequested();
                         });
+                        started.Wait();
                         var p2 = p1.Then(() => {
                         });
                         Assert.IsFalse(p1.IsResolved);
                         Assert.IsFalse(p2.IsResolved);
                         p2.Cancel();
                         Assert.IsFalse(p2.IsCancelled);
                         Assert.IsFalse(p1.IsCancelled);
                         Assert.IsTrue(requested);
                         check1.Set();
                         try {
                             p2.Join(1000);
                             Assert.Fail("The chain isn't cancelled");
                         } catch(OperationCanceledException){
                         }
                         Assert.IsTrue(p1.IsCancelled);
                         Assert.IsTrue(p2.IsCancelled);
                     }
                     [TestMethod]
                     public void CancellableAsyncTask() {
                         var finish = new Signal();
                         var started = new Signal();
                         var p = AsyncPool.RunThread(token => {
                             token.CancellationRequested(r => finish.Set());
                             started.Set();
                             finish.Wait();
                             Assert.IsTrue(token.CancelOperationIfRequested());
                         });
                         started.Wait(1000);
                         Assert.IsFalse(p.IsResolved);
                         p.Cancel();
                         try {
                             p.Join(1000);
                         } catch (OperationCanceledException) {
                         }
                     }
                 }
             }

Implab.Test/PromiseHelper.cs +2 -1

             using Implab.Parallels;
             using System.Threading;
             namespace Implab.Test {
                 static class PromiseHelper {
                     public static IPromise<T> Sleep<T>(int timeout, T retVal) {
-                        return AsyncPool.Invoke(() => {
+                        return AsyncPool.Invoke((ct) => {
+                            ct.CancellationRequested(ct.CancelOperation);
                             Thread.Sleep(timeout);
                             return retVal;
                         });
                     }
                 }
             }

Implab/ActionChainTask.cs +4 -2

             using System;
             namespace Implab {
                 public class ActionChainTask : ActionChainTaskBase, IDeferred {
                     readonly Func<IPromise> m_task;
-                    public ActionChainTask(Func<IPromise> task, Func<Exception, IPromise> error, Func<Exception, IPromise> cancel) : base(error,cancel) {
+                    public ActionChainTask(Func<IPromise> task, Func<Exception, IPromise> error, Func<Exception, IPromise> cancel, bool autoCancellable) : base(error,cancel, autoCancellable) {
                         m_task = task;
                     }
                     public void Resolve() {
                         if (m_task != null && LockCancelation()) {
                             try {
-                                Observe(m_task());
+                                var p = m_task();
+                                p.On(SetResult, HandleErrorInternal, SetCancelled);
+                                CancellationRequested(p.Cancel);
                             } catch(Exception err) {
                                 HandleErrorInternal(err);
                             }
                         }
                     }
                 }
             }

Implab/ActionChainTaskBase.cs +16 -20

             using System;
             using System.Threading;
             namespace Implab {
                 public class ActionChainTaskBase : AbstractPromise {
                     readonly Func<Exception, IPromise> m_error;
                     readonly Func<Exception, IPromise> m_cancel;
                     int m_cancelationLock;
-                    protected ActionChainTaskBase( Func<Exception, IPromise> error, Func<Exception, IPromise> cancel) {
+                    protected ActionChainTaskBase(Func<Exception, IPromise> error, Func<Exception, IPromise> cancel, bool autoCancellable) {
                         m_error = error;
                         m_cancel = cancel;
+                        if (autoCancellable)
+                            CancellationRequested(CancelOperation);
                     }
                     public void Reject(Exception error) {
                         if (LockCancelation())
                             HandleErrorInternal(error);
                     }
                     public override void CancelOperation(Exception reason) {
-                        if (m_cancel != null && LockCancelation()) {
+                        if (LockCancelation()) {
-                            try {
+                            if (m_cancel != null) {
-                                Observe(m_cancel(reason));
+                                try {
-                            } catch(Exception err) {
+                                    m_cancel(reason).On(SetResult, SetError, SetCancelled);
-                                HandleErrorInternal(err);
+                                } catch (Exception err) {
+                                    HandleErrorInternal(err);
+                                }
+                            } else {
+                                SetCancelled(reason);
                             }
                         }
                     }
                     protected void HandleErrorInternal(Exception error) {
                         if (m_error != null) {
                             try {
-                                Observe(m_error(error));
+                                var p = m_error(error);
-                            } catch(Exception err) {
+                                p.On(SetResult,SetError,SetCancelled);
+                                CancellationRequested(p.Cancel);
+                            } catch (Exception err) {
                                 SetError(err);
                             }
                         } else {
                             SetError(error);
                         }
                     }
-                    protected void Observe(IPromise operation) {
-                        if (operation == null)
-                            throw new NullReferenceException("The task returned null promise");
-                        // pass operation results to the current promise
-                        operation.On(SetResult, SetError, SetCancelled);
-                        // pass the cancelation request
-                        CancellationRequested(operation.Cancel);
                     protected bool LockCancelation() {
                         return 0 == Interlocked.CompareExchange(ref m_cancelationLock, 1, 0);
                     }
                 }
             }

Implab/ActionChainTaskT.cs +4 -2

             using System;
             namespace Implab {
                 public class ActionChainTask<T> : ActionChainTaskBase, IDeferred<T> {
                     readonly Func<T, IPromise> m_task;
-                    public ActionChainTask(Func<T, IPromise> task, Func<Exception, IPromise> error, Func<Exception, IPromise> cancel) : base(error,cancel) {
+                    public ActionChainTask(Func<T, IPromise> task, Func<Exception, IPromise> error, Func<Exception, IPromise> cancel, bool autoCancellable) : base(error,cancel, autoCancellable) {
                         m_task = task;
                     }
                     public void Resolve(T value) {
                         if (m_task != null && LockCancelation()) {
                             try {
-                                Observe(m_task(value));
+                                var p = m_task(value);
+                                p.On(SetResult, HandleErrorInternal, SetCancelled);
+                                CancellationRequested(p.Cancel);
                             } catch(Exception err) {
                                 HandleErrorInternal(err);
                             }
                         }
                     }
                 }
             }

Implab/ActionTask.cs +1 -1

             using System;
             namespace Implab {
                 public class ActionTask : ActionTaskBase, IDeferred {
                     readonly Action m_task;
-                    public ActionTask(Action task, Action<Exception> error, Action<Exception> cancel) : base(error,cancel) {
+                    public ActionTask(Action task, Action<Exception> error, Action<Exception> cancel, bool autoCancellable) : base(error,cancel, autoCancellable) {
                         m_task = task;
                     }
                     public void Resolve() {
                         if (m_task != null && LockCancelation()) {
                             try {
                                 m_task();
                                 SetResult();
                             } catch(Exception err) {
                                 HandleErrorInternal(err);
                             }
                         }
                     }
                 }
             }

Implab/ActionTaskBase.cs +3 -1

             using System;
             using System.Threading;
             namespace Implab {
                 public class ActionTaskBase : AbstractPromise {
                     readonly Action<Exception> m_cancel;
                     readonly Action<Exception> m_error;
                     int m_cancelationLock;
-                    protected ActionTaskBase( Action<Exception> error, Action<Exception> cancel) {
+                    protected ActionTaskBase( Action<Exception> error, Action<Exception> cancel, bool autoCancellable) {
                         m_error = error;
                         m_cancel = cancel;
+                        if (autoCancellable)
+                            CancellationRequested(CancelOperation);
                     }
                     public void Reject(Exception error) {
                         Safe.ArgumentNotNull(error, "error");
                         if (LockCancelation())
                             HandleErrorInternal(error);
                     }
                     protected void HandleErrorInternal(Exception error) {
                         if (m_error != null) {
                             try {
                                 m_error(error);
                                 SetResult();
                             } catch(Exception err) {
                                 SetError(err);
                             }
                         } else {
                             SetError(error);
                         }
                     }
                     public override void CancelOperation(Exception reason) {
                         if (LockCancelation()) {
                             if (m_cancel != null) {
                                 try {
                                     m_cancel(reason);
                                     SetResult();
                                 } catch (Exception err) {
                                     HandleErrorInternal(err);
                                 }
                             } else {
                                 SetCancelled(reason);
                             }
                         }
                     }
                     protected bool LockCancelation() {
                         return 0 == Interlocked.CompareExchange(ref m_cancelationLock, 1, 0);
                     }
                 }
             }

Implab/ActionTaskT.cs +1 -1

             using System;
             namespace Implab {
                 public class ActionTask<T> : ActionTaskBase, IDeferred<T> {
                     readonly Action<T> m_task;
-                    public ActionTask(Action<T> task, Action<Exception> error, Action<Exception> cancel) : base(error,cancel) {
+                    public ActionTask(Action<T> task, Action<Exception> error, Action<Exception> cancel, bool autoCancellable) : base(error,cancel, autoCancellable) {
                         m_task = task;
                     }
                     public void Resolve(T value) {
                         if (m_task != null && LockCancelation()) {
                             try {
                                 m_task(value);
                                 SetResult();
                             } catch(Exception err) {
                                 HandleErrorInternal(err);
                             }
                         }
                     }
                 }
             }

Implab/FuncChainTask.cs +5 -2

             using System;
             namespace Implab {
                 public class FuncChainTask<TResult> : FuncChainTaskBase<TResult>, IDeferred {
                     readonly Func<IPromise<TResult>> m_task;
-                    public FuncChainTask(Func<IPromise<TResult>> task, Func<Exception, IPromise<TResult>> error, Func<Exception, IPromise<TResult>> cancel) : base(error, cancel){
+                    public FuncChainTask(Func<IPromise<TResult>> task, Func<Exception, IPromise<TResult>> error, Func<Exception, IPromise<TResult>> cancel, bool autoCancellable)
+                        : base(error, cancel, autoCancellable) {
                         m_task = task;
                     }
                     public void Resolve() {
                         if (m_task != null && LockCancelation()) {
                             try {
-                                Observe(m_task());
+                                var operation = m_task();
+                                operation.On(SetResult, HandleErrorInternal, SetCancelled);
+                                CancellationRequested(operation.Cancel);
                             } catch (Exception err) {
                                 HandleErrorInternal(err);
                             }
                         }
                     }
                 }
             }

Implab/FuncChainTaskBase.cs +16 -18

             using System;
             using System.Threading;
             namespace Implab {
                 public class FuncChainTaskBase<TResult> : AbstractPromise<TResult> {
                     readonly Func<Exception, IPromise<TResult>> m_error;
                     readonly Func<Exception, IPromise<TResult>> m_cancel;
                     int m_cancelationLock;
-                    protected FuncChainTaskBase( Func<Exception, IPromise<TResult>> error, Func<Exception, IPromise<TResult>> cancel) {
+                    protected FuncChainTaskBase( Func<Exception, IPromise<TResult>> error, Func<Exception, IPromise<TResult>> cancel, bool autoCancellable) {
                         m_error = error;
                         m_cancel = cancel;
+                        if (autoCancellable)
+                            CancellationRequested(CancelOperation);
                     }
                     public void Reject(Exception error) {
                         if (LockCancelation())
                             HandleErrorInternal(error);
                     }
                     public override void CancelOperation(Exception reason) {
-                        if (m_cancel != null && LockCancelation()) {
+                        if (LockCancelation()) {
-                            try {
+                            if (m_cancel != null) {
-                                Observe(m_cancel(reason));
+                                try {
-                            } catch(Exception err) {
+                                    m_cancel(reason).On(SetResult, HandleErrorInternal, SetCancelled);
-                                HandleErrorInternal(err);
+                                } catch (Exception err) {
+                                    HandleErrorInternal(err);
+                                }
+                            } else {
+                                SetCancelled(reason);
                             }
                         }
                     }
                     protected void HandleErrorInternal(Exception error) {
                         if (m_error != null) {
                             try {
-                                Observe(m_error(error));
+                                var operation = m_error(error);
+                                operation.On(SetResult, SetError, SetCancelled);
+                                CancellationRequested(operation.Cancel);
                             } catch(Exception err) {
                                 SetError(err);
                             }
                         } else {
                             SetError(error);
                         }
                     }
-                    protected void Observe(IPromise<TResult> operation) {
-                        if (operation == null)
-                            throw new NullReferenceException("The task returned null promise");
-                        // pass operation results to the current promise
-                        operation.On(SetResult, SetError, SetCancelled);
-                        // pass the cancelation request
-                        CancellationRequested(operation.Cancel);
                     protected bool LockCancelation() {
                         return 0 == Interlocked.CompareExchange(ref m_cancelationLock, 1, 0);
                     }
                 }
             }

Implab/FuncChainTaskT.cs +4 -2

             using System;
             namespace Implab {
                 public class FuncChainTask<TArg,TResult> : FuncChainTaskBase<TResult>, IDeferred<TArg> {
                     readonly Func<TArg, IPromise<TResult>> m_task;
-                    public FuncChainTask(Func<TArg, IPromise<TResult>> task, Func<Exception, IPromise<TResult>> error, Func<Exception, IPromise<TResult>> cancel) : base(error, cancel){
+                    public FuncChainTask(Func<TArg, IPromise<TResult>> task, Func<Exception, IPromise<TResult>> error, Func<Exception, IPromise<TResult>> cancel, bool autoCancellable) : base(error, cancel, autoCancellable){
                         m_task = task;
                     }
                     public void Resolve(TArg value) {
                         if (m_task != null && LockCancelation()) {
                             try {
-                                Observe(m_task(value));
+                                var operation = m_task(value);
+                                operation.On(SetResult, HandleErrorInternal, SetCancelled);
+                                CancellationRequested(operation.Cancel);
                             } catch (Exception err) {
                                 HandleErrorInternal(err);
                             }
                         }
                     }
                 }
             }

Implab/FuncTask.cs +1 -1

             using System;
             using System.Threading;
             namespace Implab {
                 public class FuncTask<T> : FuncTaskBase<T>, IDeferred {
                     readonly Func<T> m_task;
-                    public FuncTask(Func<T> task, Func<Exception, T> error, Func<Exception, T> cancel) : base(error,cancel) {
+                    public FuncTask(Func<T> task, Func<Exception, T> error, Func<Exception, T> cancel, bool autoCancellable) : base(error, cancel, autoCancellable) {
                         m_task = task;
                     }
                     public void Resolve() {
                         if (m_task != null && LockCancelation()) {
                             try {
                                 SetResult(m_task());
                             } catch(Exception err) {
                                 HandleErrorInternal(err);
                             }
                         }
                     }
                 }
             }

Implab/FuncTaskBase.cs +3 -1

             using System;
             using System.Threading;
             namespace Implab {
                 public class FuncTaskBase<TResult> : AbstractPromise<TResult> {
                     readonly Func<Exception, TResult> m_cancel;
                     readonly Func<Exception, TResult> m_error;
                     int m_cancelationLock;
-                    protected FuncTaskBase( Func<Exception, TResult> error, Func<Exception, TResult> cancel) {
+                    protected FuncTaskBase( Func<Exception, TResult> error, Func<Exception, TResult> cancel, bool autoCancellable) {
                         m_error = error;
                         m_cancel = cancel;
+                        if (autoCancellable)
+                            CancellationRequested(CancelOperation);
                     }
                     public void Reject(Exception error) {
                         Safe.ArgumentNotNull(error, "error");
                         if (LockCancelation())
                             HandleErrorInternal(error);
                     }
                     protected void HandleErrorInternal(Exception error) {
                         if (m_error != null) {
                             try {
                                 SetResult(m_error(error));
                             } catch(Exception err) {
                                 SetError(err);
                             }
                         } else {
                             SetError(error);
                         }
                     }
                     public override void CancelOperation(Exception reason) {
                         if (LockCancelation()) {
                             if (m_cancel != null) {
                                 try {
                                     SetResult(m_cancel(reason));
                                 } catch (Exception err) {
                                     HandleErrorInternal(err);
                                 }
                             } else {
                                 SetCancelled(reason);
                             }
                         }
                     }
                     protected bool LockCancelation() {
                         return 0 == Interlocked.CompareExchange(ref m_cancelationLock, 1, 0);
                     }
                 }
             }

Implab/FuncTaskT.cs +1 -1

             using System;
             namespace Implab {
                 public class FuncTask<TArg, TResult> : FuncTaskBase<TResult>, IDeferred<TArg> {
                     readonly Func<TArg, TResult> m_task;
-                    public FuncTask(Func<TArg, TResult> task, Func<Exception, TResult> error,Func<Exception, TResult> cancel) :  base(error,cancel) {
+                    public FuncTask(Func<TArg, TResult> task, Func<Exception, TResult> error,Func<Exception, TResult> cancel, bool autoCancellable) :  base(error,cancel, autoCancellable) {
                         m_task = task;
                     }
                     public void Resolve(TArg value) {
                         if (m_task != null && LockCancelation()) {
                             try {
                                 SetResult(m_task(value));
                             } catch (Exception err) {
                                 HandleErrorInternal(err);
                             }
                         }
                     }
                 }
             }

Implab/Parallels/WorkerPool.cs +72 -2

             using System;
             using System.Threading;
             using System.Diagnostics;
             using Implab.Diagnostics;
             namespace Implab.Parallels {
                 public class WorkerPool : DispatchPool<Action> {
                     AsyncQueue<Action> m_queue = new AsyncQueue<Action>();
                     int m_queueLength;
                     readonly int m_threshold = 1;
                     public WorkerPool(int minThreads, int maxThreads, int threshold)
                         : base(minThreads, maxThreads) {
                         m_threshold = threshold;
                         InitPool();
                     }
                     public WorkerPool(int minThreads, int maxThreads) :
                         base(minThreads, maxThreads) {
                         InitPool();
                     }
                     public WorkerPool(int threads)
                         : base(threads) {
                         InitPool();
                     }
                     public WorkerPool() {
                         InitPool();
                     }
-                    public Promise<T> Invoke<T>(Func<T> task) {
+                    public IPromise<T> Invoke<T>(Func<T> task) {
+                        if (task == null)
+                            throw new ArgumentNullException("task");
+                        if (IsDisposed)
+                            throw new ObjectDisposedException(ToString());
+                        var promise = new FuncTask<T>(task, null, null, true);
+                        var lop = TraceContext.Instance.CurrentOperation;
+                        EnqueueTask(delegate {
+                            TraceContext.Instance.EnterLogicalOperation(lop, false);
+                            promise.Resolve();
+                            TraceContext.Instance.Leave();
+                        });
+                        return promise;
+                    }
+                    public IPromise Invoke(Action task) {
+                        if (task == null)
+                            throw new ArgumentNullException("task");
+                        if (IsDisposed)
+                            throw new ObjectDisposedException(ToString());
+                        var promise = new ActionTask(task, null, null, true);
+                        var lop = TraceContext.Instance.CurrentOperation;
+                        EnqueueTask(delegate {
+                            TraceContext.Instance.EnterLogicalOperation(lop, false);
+                            promise.Resolve();
+                            TraceContext.Instance.Leave();
+                        });
+                        return promise;
+                    }
+                    public IPromise<T> Invoke<T>(Func<ICancellationToken, T> task) {
                         if (task == null)
                             throw new ArgumentNullException("task");
                         if (IsDisposed)
                             throw new ObjectDisposedException(ToString());
                         var promise = new Promise<T>();
                         var lop = TraceContext.Instance.CurrentOperation;
                         EnqueueTask(delegate {
                             TraceContext.Instance.EnterLogicalOperation(lop, false);
                             try {
-                                promise.Resolve(task());
+                                if (!promise.CancelOperationIfRequested())
+                                    promise.Resolve(task(promise));
+                            } catch (Exception e) {
+                                promise.Reject(e);
+                            } finally {
+                                TraceContext.Instance.Leave();
+                            }
+                        });
+                        return promise;
+                    }
+                    public IPromise Invoke<T>(Action<ICancellationToken> task) {
+                        if (task == null)
+                            throw new ArgumentNullException("task");
+                        if (IsDisposed)
+                            throw new ObjectDisposedException(ToString());
+                        var promise = new Promise();
+                        var lop = TraceContext.Instance.CurrentOperation;
+                        EnqueueTask(delegate {
+                            TraceContext.Instance.EnterLogicalOperation(lop, false);
+                            try {
+                                if (!promise.CancelOperationIfRequested()) {
+                                    task(promise);
+                                    promise.Resolve();
+                                }
                             } catch (Exception e) {
                                 promise.Reject(e);
                             } finally {
                                 TraceContext.Instance.Leave();
                             }
                         });
                         return promise;
                     }
                     protected void EnqueueTask(Action unit) {
                         Debug.Assert(unit != null);
                         var len = Interlocked.Increment(ref m_queueLength);
                         m_queue.Enqueue(unit);
                         if (len > m_threshold * PoolSize) {
                             StartWorker();
                         }
                         SignalThread();
                     }
                     protected override bool TryDequeue(out Action unit) {
                         if (m_queue.TryDequeue(out unit)) {
                             Interlocked.Decrement(ref m_queueLength);
                             return true;
                         }
                         return false;
                     }
                     protected override void InvokeUnit(Action unit) {
                         unit();
                     }
                 }
             }

Implab/PromiseExtensions.cs +6 -6

             using System.Threading;
             using System;
             using Implab.Diagnostics;
             using System.Collections.Generic;
             #if NET_4_5
             using System.Threading.Tasks;
             #endif
             namespace Implab {
                 public static class PromiseExtensions {
                     public static IPromise<T> DispatchToCurrentContext<T>(this IPromise<T> that) {
                         Safe.ArgumentNotNull(that, "that");
                         var context = SynchronizationContext.Current;
                         if (context == null)
                             return that;
                         var p = new SyncContextPromise<T>(context);
                         p.On(that.Cancel, PromiseEventType.Cancelled);
                         that.On(
                             p.Resolve,
                             p.Reject,
                             p.Cancel
                         );
                         return p;
                     }
                     public static IPromise<T> DispatchToContext<T>(this IPromise<T> that, SynchronizationContext context) {
                         Safe.ArgumentNotNull(that, "that");
                         Safe.ArgumentNotNull(context, "context");
                         var p = new SyncContextPromise<T>(context);
                         p.On(that.Cancel, PromiseEventType.Cancelled);
                         that.On(
                             p.Resolve,
                             p.Reject,
                             p.Cancel
                         );
                         return p;
                     }
                     /// <summary>
                     /// Ensures the dispatched.
                     /// </summary>
                     /// <returns>The dispatched.</returns>
                     /// <param name="that">That.</param>
                     /// <param name="head">Head.</param>
                     /// <param name="cleanup">Cleanup.</param>
                     /// <typeparam name="TPromise">The 1st type parameter.</typeparam>
                     /// <typeparam name="T">The 2nd type parameter.</typeparam>
                     public static TPromise EnsureDispatched<TPromise,T>(this TPromise that, IPromise<T> head, Action<T> cleanup) where TPromise : IPromise{
                         Safe.ArgumentNotNull(that, "that");
                         Safe.ArgumentNotNull(head, "head");
                         that.On(() => head.On(cleanup), PromiseEventType.Cancelled);
                         return that;
                     }
                     public static AsyncCallback AsyncCallback<T>(this Promise<T> that, Func<IAsyncResult,T> callback) {
                         Safe.ArgumentNotNull(that, "that");
                         Safe.ArgumentNotNull(callback, "callback");
                         var op = TraceContext.Instance.CurrentOperation;
                         return ar => {
                             TraceContext.Instance.EnterLogicalOperation(op,false);
                             try {
                                 that.Resolve(callback(ar));
                             } catch (Exception err) {
                                 that.Reject(err);
                             } finally {
                                 TraceContext.Instance.Leave();
                             }
                         };
                     }
                     static void CancelCallback(object cookie) {
                         ((ICancellable)cookie).Cancel();
                     }
                     /// <summary>
                     /// Cancells promise after the specified timeout is elapsed.
                     /// </summary>
                     /// <param name="that">The promise to cancel on timeout.</param>
                     /// <param name="milliseconds">The timeout in milliseconds.</param>
                     /// <typeparam name="TPromise">The 1st type parameter.</typeparam>
                     public static TPromise Timeout<TPromise>(this TPromise that, int milliseconds) where TPromise : IPromise {
                         Safe.ArgumentNotNull(that, "that");
                         var timer = new Timer(CancelCallback, that, milliseconds, -1);
                         that.On(timer.Dispose, PromiseEventType.All);
                         return that;
                     }
                     public static IPromise Bundle(this ICollection<IPromise> that) {
                         Safe.ArgumentNotNull(that, "that");
                         int count = that.Count;
                         int errors = 0;
                         var medium = new Promise();
                         if (count == 0) {
                             medium.Resolve();
                             return medium;
                         }
                         medium.On(() => {
                             foreach(var p2 in that)
                                 p2.Cancel();
                         }, PromiseEventType.ErrorOrCancel);
                         foreach (var p in that)
                             p.On(
                                 () => {
                                     if (Interlocked.Decrement(ref count) == 0)
                                         medium.Resolve();
                                 },
                                 error => {
                                     if (Interlocked.Increment(ref errors) == 1)
                                         medium.Reject(
                                             new Exception("The dependency promise is failed", error)
                                         );
                                 },
                                 reason => {
                                     if (Interlocked.Increment(ref errors) == 1)
                                         medium.Cancel(
                                             new Exception("The dependency promise is cancelled")
                                         );
                                 }
                             );
                         return medium;
                     }
                     public static IPromise<T[]> Bundle<T>(this ICollection<IPromise<T>> that) {
                         Safe.ArgumentNotNull(that, "that");
                         int count = that.Count;
                         int errors = 0;
                         var medium = new Promise<T[]>();
                         var results = new T[that.Count];
                         medium.On(() => {
                             foreach(var p2 in that)
                                 p2.Cancel();
                         }, PromiseEventType.ErrorOrCancel);
                         int i = 0;
                         foreach (var p in that) {
                             var idx = i;
                             p.On(
                                 x => {
                                     results[idx] = x;
                                     if (Interlocked.Decrement(ref count) == 0)
                                         medium.Resolve(results);
                                 },
                                 error => {
                                     if (Interlocked.Increment(ref errors) == 1)
                                         medium.Reject(
                                             new Exception("The dependency promise is failed", error)
                                         );
                                 },
                                 reason => {
                                     if (Interlocked.Increment(ref errors) == 1)
                                         medium.Cancel(
                                             new Exception("The dependency promise is cancelled", reason)
                                         );
                                 }
                             );
                             i++;
                         }
                         return medium;
                     }
                     public static IPromise Then(this IPromise that, Action success, Action<Exception> error, Action<Exception> cancel) {
                         Safe.ArgumentNotNull(that, "that");
-                        var d = new ActionTask(success, error, cancel);
+                        var d = new ActionTask(success, error, cancel, false);
                         that.On(d.Resolve, d.Reject, d.CancelOperation);
                         if (success != null)
                             d.CancellationRequested(that.Cancel);
                         return d;
                     }
                     public static IPromise Then(this IPromise that, Action success, Action<Exception> error) {
                         return Then(that, success, error, null);
                     }
                     public static IPromise Then(this IPromise that, Action success) {
                         return Then(that, success, null, null);
                     }
                     public static IPromise<T> Then<T>(this IPromise that, Func<T> success, Func<Exception, T> error, Func<Exception, T> cancel) {
                         Safe.ArgumentNotNull(that, "that");
-                        var d = new FuncTask<T>(success, error, cancel);
+                        var d = new FuncTask<T>(success, error, cancel, false);
                         that.On(d.Resolve, d.Reject, d.CancelOperation);
                         if (success != null)
                             d.CancellationRequested(that.Cancel);
                         return d;
                     }
                     public static IPromise<T> Then<T>(this IPromise that, Func<T> success, Func<Exception, T> error) {
                         return Then(that, success, error, null);
                     }
                     public static IPromise<T> Then<T>(this IPromise that, Func<T> success) {
                         return Then(that, success, null, null);
                     }
                     public static IPromise<T2> Then<T, T2>(this IPromise<T> that, Func<T, T2> success, Func<Exception, T2> error, Func<Exception, T2> cancel) {
                         Safe.ArgumentNotNull(that, "that");
-                        var d = new FuncTask<T,T2>(success, error, cancel);
+                        var d = new FuncTask<T,T2>(success, error, cancel, false);
                         that.On(d.Resolve, d.Reject, d.CancelOperation);
                         if (success != null)
                             d.CancellationRequested(that.Cancel);
                         return d;
                     }
                     public static IPromise<T2> Then<T, T2>(this IPromise<T> that, Func<T, T2> success, Func<Exception, T2> error) {
                         return Then(that, success, error, null);
                     }
                     public static IPromise<T2> Then<T, T2>(this IPromise<T> that, Func<T, T2> success) {
                         return Then(that, success, null, null);
                     }
                     #region chain traits
                     public static IPromise Chain(this IPromise that, Func<IPromise> success, Func<Exception,IPromise> error, Func<Exception,IPromise> cancel) {
                         Safe.ArgumentNotNull(that, "that");
-                        var d = new ActionChainTask(success, error, cancel);
+                        var d = new ActionChainTask(success, error, cancel, false);
                         that.On(d.Resolve, d.Reject, d.CancelOperation);
                         if (success != null)
                             d.CancellationRequested(that.Cancel);
                         return d;
                     }
                     public static IPromise Chain(this IPromise that, Func<IPromise> success, Func<Exception,IPromise> error) {
                         return Chain(that, success, error, null);
                     }
                     public static IPromise Chain(this IPromise that, Func<IPromise> success) {
                         return Chain(that, success, null, null);
                     }
                     public static IPromise<T> Chain<T>(this IPromise that, Func<IPromise<T>> success, Func<Exception, IPromise<T>> error, Func<Exception, IPromise<T>> cancel) {
                         Safe.ArgumentNotNull(that, "that");
-                        var d = new FuncChainTask<T>(success, error, cancel);
+                        var d = new FuncChainTask<T>(success, error, cancel, false);
                         that.On(d.Resolve, d.Reject, d.CancelOperation);
                         if (success != null)
                             d.CancellationRequested(that.Cancel);
                         return d;
                     }
                     public static IPromise<T> Chain<T>(this IPromise that, Func<IPromise<T>> success, Func<Exception, IPromise<T>> error) {
                         return Chain(that, success, error, null);
                     }
                     public static IPromise<T> Chain<T>(this IPromise that, Func<IPromise<T>> success) {
                         return Chain(that, success, null, null);
                     }
                     public static IPromise<T2> Chain<T, T2>(this IPromise<T> that, Func<T, IPromise<T2>> success, Func<Exception, IPromise<T2>> error, Func<Exception, IPromise<T2>> cancel) {
                         Safe.ArgumentNotNull(that, "that");
-                        var d = new FuncChainTask<T,T2>(success, error, cancel);
+                        var d = new FuncChainTask<T,T2>(success, error, cancel, false);
                         that.On(d.Resolve, d.Reject, d.CancelOperation);
                         if (success != null)
                             d.CancellationRequested(that.Cancel);
                         return d;
                     }
                     public static IPromise<T2> Chain<T, T2>(this IPromise<T> that, Func<T, IPromise<T2>> success, Func<Exception, IPromise<T2>> error) {
                         return Chain(that, success, error, null);
                     }
                     public static IPromise<T2> Chain<T, T2>(this IPromise<T> that, Func<T, IPromise<T2>> success) {
                         return Chain(that, success, null, null);
                     }
                     #endregion
                     #if NET_4_5
                     public static Task<T> GetTask<T>(this IPromise<T> that) {
                         Safe.ArgumentNotNull(that, "that");
                         var tcs = new TaskCompletionSource<T>();
                         that.On(tcs.SetResult, tcs.SetException, r => tcs.SetCanceled());
                         return tcs.Task;
                     }
                     #endif
                 }
             }

General Comments 0

Write
Preview

You need to be logged in to leave comments. Login now

No TODOs yet

	Site-wide shortcuts
/	Use quick search box
g h	Goto home page
g g	Goto my private gists page
g G	Goto my public gists page
g 0-9	Goto bookmarked items from 0-9
n r	New repository page
n g	New gist page

	Repositories
g s	Goto summary page
g c	Goto changelog page
g f	Goto files page
g F	Goto files page with file search activated
g p	Goto pull requests page
g o	Goto repository settings
g O	Goto repository access permissions settings
t s	Toggle sidebar on some pages