implab/ImplabNet Commit - r124:a336cb13c6a9

major update, added Drain mathod to AsyncQueue class

cin -

r124:a336cb13c6a9 v2

parent child

Context file:

r124:a336cb13c6a9

Collapse all files

Implab.Test/AsyncTests.cs +108 -5

             using System;
             using System.Reflection;
             using System.Threading;
             using Implab.Parallels;
             #if MONO
             using NUnit.Framework;
             using TestClassAttribute = NUnit.Framework.TestFixtureAttribute;
             using TestMethod = 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.Cancel();
                         var p2 = p.Cancelled(() => {
                             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.Cancel();
                         var p2 = p
                             .Cancelled<bool>(() => {
                                 throw new ApplicationException("CANCELLED");
                             })
                             .Error(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.Error(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
-                                .InvokeNewThread(() => {
+                                .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
-                                .InvokeNewThread(() => {
+                                .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);
                             }
                         )
-                            .Combine()
+                            .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);
                             }
                         )
-                            .Combine()
+                            .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);
                             }
                         )
-                            .Combine()
+                            .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(0,10,1)) {
                             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 p = PromiseHelper
                             .Sleep(1, "Hi, HAL!")
                             .Chain(x => {
                                 hemStarted.Set();
                                 // запускаем две асинхронные операции
                                 var result = PromiseHelper
                                     .Sleep(100000000, "HEM ENABLED!!!")
                                     .Then(s => {
                                         pSurvive.Resolve(false);
                                         return s;
                                     });
                                 result
                                     .Cancelled(() => pSurvive.Resolve(true));
                                 return result;
                             });
                         hemStarted.WaitOne();
                         p.Cancel();
                         try {
                             p.Join();
                         } catch (OperationCanceledException) {
                             Assert.IsTrue(pSurvive.Join());
                         }
                     }
                 }
             }

Implab/Parallels/ArrayTraits.cs +1 -1

             using Implab.Diagnostics;
             using System;
             using System.Diagnostics;
             using System.Threading;
             namespace Implab.Parallels {
                 public static class ArrayTraits {
                     class ArrayIterator<TSrc> : DispatchPool<int> {
                         readonly Action<TSrc> m_action;
                         readonly TSrc[] m_source;
                         readonly Promise<int> m_promise = new Promise<int>();
                         readonly LogicalOperation m_logicalOperation;
                         int m_pending;
                         int m_next;
                         public ArrayIterator(TSrc[] source, Action<TSrc> action, int threads)
                             : base(threads) {
                             Debug.Assert(source != null);
                             Debug.Assert(action != null);
                             m_logicalOperation = TraceContext.Instance.CurrentOperation;
                             m_next = 0;
                             m_source = source;
                             m_pending = source.Length;
                             m_action = action;
                             m_promise.On(Dispose, PromiseEventType.All);
                             InitPool();
                         }
                         public Promise<int> Promise {
                             get {
                                 return m_promise;
                             }
                         }
                         protected override void Worker() {
                             TraceContext.Instance.EnterLogicalOperation(m_logicalOperation, false);
                             try {
                                 base.Worker();
                             } finally {
                                 TraceContext.Instance.Leave();
                             }
                         }
                         protected override bool TryDequeue(out int unit) {
                             unit = Interlocked.Increment(ref m_next) - 1;
                             return unit < m_source.Length;
                         }
                         protected override void InvokeUnit(int unit) {
                             try {
                                 m_action(m_source[unit]);
                                 var pending = Interlocked.Decrement(ref m_pending);
                                 if (pending == 0)
                                     m_promise.Resolve(m_source.Length);
                             } catch (Exception e) {
                                 m_promise.Reject(e);
                             }
                         }
                     }
                     class ArrayMapper<TSrc, TDst>: DispatchPool<int> {
                         readonly Func<TSrc, TDst> m_transform;
                         readonly TSrc[] m_source;
                         readonly TDst[] m_dest;
                         readonly Promise<TDst[]> m_promise = new Promise<TDst[]>();
                         readonly LogicalOperation m_logicalOperation;
                         int m_pending;
                         int m_next;
                         public ArrayMapper(TSrc[] source, Func<TSrc, TDst> transform, int threads)
                             : base(threads) {
                             Debug.Assert (source != null);
                             Debug.Assert( transform != null);
                             m_next = 0;
                             m_source = source;
                             m_dest = new TDst[source.Length];
                             m_pending = source.Length;
                             m_transform = transform;
                             m_logicalOperation = TraceContext.Instance.CurrentOperation;
                             m_promise.On(Dispose, PromiseEventType.All);
                             InitPool();
                         }
                         public Promise<TDst[]> Promise {
                             get {
                                 return m_promise;
                             }
                         }
                         protected override void Worker() {
                             TraceContext.Instance.EnterLogicalOperation(m_logicalOperation,false);
                             try {
                                 base.Worker();
                             } finally {
                                 TraceContext.Instance.Leave();
                             }
                         }
                         protected override bool TryDequeue(out int unit) {
                             unit = Interlocked.Increment(ref m_next) - 1;
                             return unit < m_source.Length;
                         }
                         protected override void InvokeUnit(int unit) {
                             try {
                                 m_dest[unit] = m_transform(m_source[unit]);
                                 var pending = Interlocked.Decrement(ref m_pending);
                                 if (pending == 0)
                                     m_promise.Resolve(m_dest);
                             } catch (Exception e) {
                                 m_promise.Reject(e);
                             }
                         }
                     }
                     public static IPromise<TDst[]> ParallelMap<TSrc, TDst> (this TSrc[] source, Func<TSrc,TDst> transform, int threads) {
                         if (source == null)
                             throw new ArgumentNullException("source");
                         if (transform == null)
                             throw new ArgumentNullException("transform");
                         var mapper = new ArrayMapper<TSrc, TDst>(source, transform, threads);
                         return mapper.Promise;
                     }
                     public static IPromise<int> ParallelForEach<TSrc>(this TSrc[] source, Action<TSrc> action, int threads) {
                         if (source == null)
                             throw new ArgumentNullException("source");
                         if (action == null)
                             throw new ArgumentNullException("action");
                         var iter = new ArrayIterator<TSrc>(source, action, threads);
                         return iter.Promise;
                     }
                     public static IPromise<TDst[]> ChainedMap<TSrc, TDst>(this TSrc[] source, Func<TSrc, IPromise<TDst>> transform, int threads) {
                         if (source == null)
                             throw new ArgumentNullException("source");
                         if (transform == null)
                             throw new ArgumentNullException("transform");
                         if (threads <= 0)
                             throw new ArgumentOutOfRangeException("threads","Threads number must be greater then zero");
                         if (source.Length == 0)
                             return Promise<TDst[]>.ResultToPromise(new TDst[0]);
                         var promise = new Promise<TDst[]>();
                         var res = new TDst[source.Length];
                         var pending = source.Length;
                         object locker = new object();
                         int slots = threads;
                         // Analysis disable AccessToDisposedClosure
-                        AsyncPool.InvokeNewThread<int>(() => {
+                        AsyncPool.RunThread<int>(() => {
                             for (int i = 0; i < source.Length; i++) {
                                 if(promise.IsResolved)
                                     break; // stop processing in case of error or cancellation
                                 var idx = i;
                                 if (Interlocked.Decrement(ref slots) < 0) {
                                     lock(locker) {
                                         while(slots < 0)
                                             Monitor.Wait(locker);
                                     }
                                 }
                                 try {
                                     transform(source[i])
                                         .On( x => {
                                             Interlocked.Increment(ref slots);
                                             lock (locker) {
                                                 Monitor.Pulse(locker);
                                             }
                                         })
                                         .On(
                                             x => {
                                                 res[idx] = x;
                                                 var left = Interlocked.Decrement(ref pending);
                                                 if (left == 0)
                                                     promise.Resolve(res);
                                             },
                                             promise.Reject
                                         );
                                 } catch (Exception e) {
                                     promise.Reject(e);
                                 }
                             }
                             return 0;
                         });
                         return promise;
                     }
                 }
             }

Implab/Parallels/AsyncPool.cs +4 -4

             using Implab.Diagnostics;
             using System;
             using System.Threading;
             using System.Linq;
             namespace Implab.Parallels {
             	/// <summary>
             	/// Класс для распаралеливания задач.
             	/// </summary>
             	/// <remarks>
             	/// Используя данный класс и лямда выражения можно распараллелить
             	/// вычисления, для этого используется концепция обещаний.
             	/// </remarks>
             	public static class AsyncPool {
             		public static IPromise<T> Invoke<T>(Func<T> func) {
             			var p = new Promise<T>();
                         var caller = TraceContext.Instance.CurrentOperation;
             			ThreadPool.QueueUserWorkItem(param => {
                             TraceContext.Instance.EnterLogicalOperation(caller,false);
             				try {
             					p.Resolve(func());
             				} catch(Exception e) {
             					p.Reject(e);
                             } finally {
                                 TraceContext.Instance.Leave();
                             }
             			});
             			return p;
             		}
-                    public static IPromise<T> InvokeNewThread<T>(Func<T> func) {
+                    public static IPromise<T> RunThread<T>(Func<T> func) {
                         var p = new Promise<T>();
                         var caller = TraceContext.Instance.CurrentOperation;
                         var worker = new Thread(() => {
                             TraceContext.Instance.EnterLogicalOperation(caller,false);
                             try {
                                 p.Resolve(func());
                             } catch (Exception e) {
                                 p.Reject(e);
                             } finally {
                                 TraceContext.Instance.Leave();
                             }
                         });
                         worker.IsBackground = true;
                         worker.Start();
                         return p;
                     }
-                    public static IPromise InvokeNewThread(Action func) {
+                    public static IPromise RunThread(Action func) {
                         var p = new Promise();
                         var caller = TraceContext.Instance.CurrentOperation;
                         var worker = new Thread(() => {
                             TraceContext.Instance.EnterLogicalOperation(caller,false);
                             try {
                                 func();
                                 p.Resolve();
                             } catch (Exception e) {
                                 p.Reject(e);
                             } finally {
                                 TraceContext.Instance.Leave();
                             }
                         });
                         worker.IsBackground = true;
                         worker.Start();
                         return p;
                     }
                     public static IPromise[] RunThread(params Action[] func) {
-                        return func.Select(f => InvokeNewThread(f)).ToArray();
+                        return func.Select(f => RunThread(f)).ToArray();
                     }
                     public static IPromise<T>[] RunThread<T>(params Func<T>[] func) {
-                        return func.Select(f => InvokeNewThread(f)).ToArray();
+                        return func.Select(f => RunThread(f)).ToArray();
                     }
             	}
             }

Implab/Parallels/AsyncQueue.cs +76 -27

             using System.Threading;
             using System.Collections.Generic;
             using System;
             using System.Collections;
+            using System.Diagnostics;
             namespace Implab.Parallels {
                 public class AsyncQueue<T> : IEnumerable<T> {
                     class Chunk {
                         public Chunk next;
                         int m_low;
                         int m_hi;
                         int m_alloc;
                         readonly int m_size;
                         readonly T[] m_data;
                         public Chunk(int size) {
                             m_size = size;
                             m_data = new T[size];
                         }
                         public Chunk(int size, T value) {
                             m_size = size;
                             m_hi = 1;
                             m_alloc = 1;
                             m_data = new T[size];
                             m_data[0] = value;
                         }
                         public Chunk(int size, T[] data, int offset, int length, int alloc) {
                             m_size = size;
                             m_hi = length;
                             m_alloc = alloc;
                             m_data = new T[size];
                             Array.Copy(data, offset, m_data, 0, length);
                         }
                         public int Low {
                             get { return m_low; }
                         }
                         public int Hi {
                             get { return m_hi; }
                         }
                         public bool TryEnqueue(T value, out bool extend) {
                             var alloc = Interlocked.Increment(ref m_alloc) - 1;
                             if (alloc >= m_size) {
                                 extend = alloc == m_size;
                                 return false;
                             }
                             extend = false;
                             m_data[alloc] = value;
                             while (alloc != Interlocked.CompareExchange(ref m_hi, alloc + 1, alloc)) {
                                 // spin wait for commit
                             }
                             return true;
                         }
+                        /// <summary>
+                        /// Prevents from allocating new space in the chunk and waits for all write operations to complete
+                        /// </summary>
+                        public void Commit() {
+                            var actual = Math.Min(Interlocked.Exchange(ref m_alloc, m_size + 1), m_size);
+                            while (m_hi != actual)
+                                Thread.MemoryBarrier();
+                        }
                         public bool TryDequeue(out T value, out bool recycle) {
                             int low;
                             do {
                                 low = m_low;
                                 if (low >= m_hi) {
                                     value = default(T);
                                     recycle = (low == m_size);
                                     return false;
                                 }
                             } while(low != Interlocked.CompareExchange(ref m_low, low + 1, low));
                             recycle = (low == m_size - 1);
                             value = m_data[low];
                             return true;
                         }
                         public bool TryEnqueueBatch(T[] batch, int offset, int length, out int enqueued, out bool extend) {
                             //int alloc;
                             //int allocSize;
                             var alloc = Interlocked.Add(ref m_alloc, length) - length;
                             if (alloc > m_size) {
                                 // the chunk is full and someone already
                                 // creating the new one
                                 enqueued = 0; // nothing was added
                                 extend = false; // the caller shouldn't try to extend the queue
                                 return false; // nothing was added
                             }
                             enqueued = Math.Min(m_size - alloc, length);
                             extend = length > enqueued;
                             if (enqueued == 0)
                                 return false;
                             Array.Copy(batch, offset, m_data, alloc, enqueued);
                             while (alloc != Interlocked.CompareExchange(ref m_hi, alloc + enqueued, alloc)) {
                                 // spin wait for commit
                             }
                             return true;
                         }
                         public bool TryDequeueBatch(T[] buffer, int offset, int length,out int dequeued, out bool recycle) {
                             int low, hi, batchSize;
                             do {
                                 low = m_low;
                                 hi = m_hi;
                                 if (low >= hi) {
                                     dequeued = 0;
                                     recycle = (low == m_size); // recycling could be restarted and we need to signal again
                                     return false;
                                 }
                                 batchSize = Math.Min(hi - low, length);
                             } while(low != Interlocked.CompareExchange(ref m_low, low + batchSize, low));
                             recycle = (low == m_size - batchSize);
                             dequeued = batchSize;
                             Array.Copy(m_data, low, buffer, offset, batchSize);
                             return true;
                         }
                         public T GetAt(int pos) {
                             return m_data[pos];
                         }
                     }
                     public const int DEFAULT_CHUNK_SIZE = 32;
                     public const int MAX_CHUNK_SIZE = 262144;
                     readonly int m_chunkSize = DEFAULT_CHUNK_SIZE;
                     Chunk m_first;
                     Chunk m_last;
                     public AsyncQueue() {
                         m_last = m_first = new Chunk(m_chunkSize);
                     }
                     /// <summary>
                     /// Adds the specified value to the queue.
                     /// </summary>
                     /// <param name="value">Tha value which will be added to the queue.</param>
                     public void Enqueue(T value) {
                         var last = m_last;
                         // spin wait to the new chunk
                         bool extend = true;
                         while (last == null || !last.TryEnqueue(value, out extend)) {
                             // try to extend queue
                             if (extend || last == null) {
                                 var chunk = new Chunk(m_chunkSize, value);
                                 if (EnqueueChunk(last, chunk))
                                     break; // success! exit!
                                 last = m_last;
                             } else {
                                 while (last == m_last) {
                                     Thread.MemoryBarrier();
                                 }
                                 last = m_last;
                             }
                         }
                     }
                     /// <summary>
                     /// Adds the specified data to the queue.
                     /// </summary>
                     /// <param name="data">The buffer which contains the data to be enqueued.</param>
                     /// <param name="offset">The offset of the data in the buffer.</param>
                     /// <param name="length">The size of the data to read from the buffer.</param>
                     public void EnqueueRange(T[] data, int offset, int length) {
                         if (data == null)
                             throw new ArgumentNullException("data");
                         if (offset < 0)
                             throw new ArgumentOutOfRangeException("offset");
                         if (length < 1 || offset + length > data.Length)
                             throw new ArgumentOutOfRangeException("length");
                         var last = m_last;
                         bool extend;
                         int enqueued;
                         while (length > 0) {
                             extend = true;
                             if (last != null && last.TryEnqueueBatch(data, offset, length, out enqueued, out extend)) {
                                 length -= enqueued;
                                 offset += enqueued;
                             }
                             if (extend) {
                                 // there was no enough space in the chunk
                                 // or there was no chunks in the queue
                                 while (length > 0) {
                                     var size = Math.Min(length, MAX_CHUNK_SIZE);
                                     var chunk = new Chunk(
                                         Math.Max(size, m_chunkSize),
                                         data,
                                         offset,
                                         size,
                                         length // length >= size
                                     );
                                     if (!EnqueueChunk(last, chunk)) {
                                         // looks like the queue has been updated then proceed from the beginning
                                         last = m_last;
                                         break;
                                     }
                                     // we have successfully added the new chunk
                                     last = chunk;
                                     length -= size;
                                     offset += size;
                                 }
                             } else {
                                 // we don't need to extend the queue, if we successfully enqueued data
                                 if (length == 0)
                                     break;
                                 // if we need to wait while someone is extending the queue
                                 // spinwait
                                 while (last == m_last) {
                                     Thread.MemoryBarrier();
                                 }
                                 last = m_last;
                             }
                         }
                     }
                     /// <summary>
                     /// Tries to retrieve the first element from the queue.
                     /// </summary>
                     /// <returns><c>true</c>, if element is dequeued, <c>false</c> otherwise.</returns>
                     /// <param name="value">The value of the dequeued element.</param>
                     public bool TryDequeue(out T value) {
                         var chunk = m_first;
                         bool recycle;
                         while (chunk != null) {
                             var result = chunk.TryDequeue(out value, out recycle);
                             if (recycle) // this chunk is waste
                                 RecycleFirstChunk(chunk);
                             else
                                 return result; // this chunk is usable and returned actual result
                             if (result) // this chunk is waste but the true result is always actual
                                 return true;
                             // try again
                             chunk = m_first;
                         }
                         // the queue is empty
                         value = default(T);
                         return false;
                     }
                     /// <summary>
                     /// Tries to dequeue the specified amount of data from the queue.
                     /// </summary>
                     /// <returns><c>true</c>, if data was deuqueued, <c>false</c> otherwise.</returns>
                     /// <param name="buffer">The buffer to which the data will be written.</param>
                     /// <param name="offset">The offset in the buffer at which the data will be written.</param>
                     /// <param name="length">The maximum amount of data to be retrieved.</param>
                     /// <param name="dequeued">The actual amout of the retrieved data.</param>
                     public bool TryDequeueRange(T[] buffer, int offset, int length, out int dequeued) {
                         if (buffer == null)
                             throw new ArgumentNullException("buffer");
                         if (offset < 0)
                             throw new ArgumentOutOfRangeException("offset");
                         if (length < 1 || offset + length > buffer.Length)
                             throw new ArgumentOutOfRangeException("length");
                         var chunk = m_first;
                         bool recycle;
                         dequeued = 0;
                         while (chunk != null) {
                             int actual;
                             if (chunk.TryDequeueBatch(buffer, offset, length, out actual, out recycle)) {
                                 offset += actual;
                                 length -= actual;
                                 dequeued += actual;
                             }
                             if (recycle) // this chunk is waste
                                 RecycleFirstChunk(chunk);
                             else if (actual == 0)
                                 break; // the chunk is usable, but doesn't contain any data (it's the last chunk in the queue)
                             if (length == 0)
                                 return true;
                             // we still may dequeue something
                             // try again
                             chunk = m_first;
                         }
                         return dequeued != 0;
                     }
                     /// <summary>
                     /// Tries to dequeue all remaining data in the first chunk.
                     /// </summary>
                     /// <returns><c>true</c>, if data was dequeued, <c>false</c> otherwise.</returns>
                     /// <param name="buffer">The buffer to which the data will be written.</param>
                     /// <param name="offset">The offset in the buffer at which the data will be written.</param>
                     /// <param name="length">Tha maximum amount of the data to be dequeued.</param>
                     /// <param name="dequeued">The actual amount of the dequeued data.</param>
                     public bool TryDequeueChunk(T[] buffer, int offset, int length, out int dequeued) {
                         if (buffer == null)
                             throw new ArgumentNullException("buffer");
                         if (offset < 0)
                             throw new ArgumentOutOfRangeException("offset");
                         if (length < 1 || offset + length > buffer.Length)
                             throw new ArgumentOutOfRangeException("length");
                         var chunk = m_first;
                         bool recycle;
                         dequeued = 0;
                         while (chunk != null) {
                             int actual;
                             if (chunk.TryDequeueBatch(buffer, offset, length, out actual, out recycle)) {
                                 dequeued = actual;
                             }
                             if (recycle) // this chunk is waste
                                 RecycleFirstChunk(chunk);
                             // if we have dequeued any data, then return
                             if (dequeued != 0)
                                 return true;
                             // we still may dequeue something
                             // try again
                             chunk = m_first;
                         }
                         return false;
                     }
                     bool EnqueueChunk(Chunk last, Chunk chunk) {
                         if (Interlocked.CompareExchange(ref m_last, chunk, last) != last)
                             return false;
                         if (last != null)
                             last.next = chunk;
-                        else
+                        else {
                             m_first = chunk;
+                        }
                         return true;
                     }
                     void RecycleFirstChunk(Chunk first) {
                         var next = first.next;
+                        if (first != Interlocked.CompareExchange(ref m_first, next, first))
+                            return;
                         if (next == null) {
-                            // looks like this is the last chunk
                             if (first != Interlocked.CompareExchange(ref m_last, null, first)) {
+                                /*while (first.next == null)
+                                    Thread.MemoryBarrier();*/
                                 // race
-                                // maybe someone already recycled this chunk
+                                // someone already updated the tail, restore the pointer to the queue head
-                                // or a new chunk has been appedned to the queue
+                                m_first = first;
-                                return; // give up
                             }
                             // the tail is updated
                         }
                         // we need to update the head
-                        Interlocked.CompareExchange(ref m_first, next, first);
+                        //Interlocked.CompareExchange(ref m_first, next, first);
                         // if the head is already updated then give up
-                        return;
+                        //return;
                     }
                     public void Clear() {
                         // start the new queue
-                        var t = new Chunk(m_chunkSize);
+                        var chunk = new Chunk(m_chunkSize);
-                        Thread.MemoryBarrier();
-                        m_last = t;
+                        do {
-                        Thread.MemoryBarrier();
+                            Thread.MemoryBarrier();
+                            var first = m_first;
+                            var last = m_last;
+                            if (last == null) // nothing to clear
+                                return;
-                        // make the new queue available to the readers, and stop the old one
+                            if (first == null || (first.next == null && first != last)) // inconcistency
-                        m_first = t;
+                                continue;
-                        Thread.MemoryBarrier();
+                            // here we will create inconsistency which will force others to spin
+                            // and prevent from fetching. chunk.next = null
+                            if (first != Interlocked.CompareExchange(ref m_first, chunk, first))
+                                continue;// inconsistent
+                            m_last = chunk;
+                            return;
+                        } while(true);
                     }
                     public T[] Drain() {
                         // start the new queue
-                        var t = new Chunk(m_chunkSize);
+                        var chunk = new Chunk(m_chunkSize);
-                        Thread.MemoryBarrier();
-                        m_last = t;
-                        Thread.MemoryBarrier();
-                        // make the new queue available to the readers, and stop the old one
-                        Chunk first;
                         do {
-                            first = m_first;
+                            Thread.MemoryBarrier();
-                        } while(first != Interlocked.CompareExchange(ref m_first
+                            var first = m_first;
-                        Thread.MemoryBarrier();
+                            var last = m_last;
+                            if (last == null)
+                                return new T[0];
+                            if (first == null || (first.next == null && first != last))
+                                continue;
+                            // here we will create inconsistency which will force others to spin
+                            // and prevent from fetching. chunk.next = null
+                            if (first != Interlocked.CompareExchange(ref m_first, chunk, first))
+                                continue;// inconsistent
+                            last = Interlocked.Exchange(ref m_last, chunk);
+                            return ReadChunks(first, last);
+                        } while(true);
                     }
-                    T[] ReadChunks(Chunk chunk) {
+                    T[] ReadChunks(Chunk chunk, object last) {
                         var result = new List<T>();
                         var buffer = new T[m_chunkSize];
                         int actual;
                         bool recycle;
                         while (chunk != null) {
+                            // ensure all write operations on the chunk are complete
+                            chunk.Commit();
                             // we need to read the chunk using this way
                             // since some client still may completing the dequeue
                             // operation, such clients most likely won't get results
                             while (chunk.TryDequeueBatch(buffer, 0, buffer.Length, out actual, out recycle))
                                 result.AddRange(new ArraySegmentCollection(buffer, 0, actual));
-                            chunk = chunk.next;
+                            if (chunk == last) {
+                                chunk = null;
+                            } else {
+                                while (chunk.next == null)
+                                    Thread.MemoryBarrier();
+                                chunk = chunk.next;
+                            }
                         }
                         return result.ToArray();
                     }
                     struct ArraySegmentCollection : ICollection<T> {
                         readonly T[] m_data;
                         readonly int m_offset;
                         readonly int m_length;
                         public ArraySegmentCollection(T[] data, int offset, int length) {
                             m_data = data;
                             m_offset = offset;
                             m_length = length;
                         }
                         #region ICollection implementation
                         public void Add(T item) {
                             throw new InvalidOperationException();
                         }
                         public void Clear() {
                             throw new InvalidOperationException();
                         }
                         public bool Contains(T item) {
                             return false;
                         }
                         public void CopyTo(T[] array, int arrayIndex) {
                             Array.Copy(m_data,m_offset,array,arrayIndex, m_length);
                         }
                         public bool Remove(T item) {
                             throw new NotImplementedException();
                         }
                         public int Count {
                             get {
                                 return m_length;
                             }
                         }
                         public bool IsReadOnly {
                             get {
                                 return true;
                             }
                         }
                         #endregion
                         #region IEnumerable implementation
                         public IEnumerator<T> GetEnumerator() {
                             for (int i = m_offset; i < m_length + m_offset; i++)
                                 yield return m_data[i];
                         }
                         #endregion
                         #region IEnumerable implementation
                         IEnumerator IEnumerable.GetEnumerator() {
                             return GetEnumerator();
                         }
                         #endregion
                     }
                     #region IEnumerable implementation
                     class Enumerator : IEnumerator<T> {
                         Chunk m_current;
                         int m_pos = -1;
                         public Enumerator(Chunk fisrt) {
                             m_current = fisrt;
                         }
                         #region IEnumerator implementation
                         public bool MoveNext() {
                             if (m_current == null)
                                 return false;
                             if (m_pos == -1)
                                 m_pos = m_current.Low;
                             else
                                 m_pos++;
                             if (m_pos == m_current.Hi) {
                                 m_pos = 0;
                                 m_current = m_current.next;
                             }
                             return true;
                         }
                         public void Reset() {
                             throw new NotSupportedException();
                         }
                         object IEnumerator.Current {
                             get {
                                 return Current;
                             }
                         }
                         #endregion
                         #region IDisposable implementation
                         public void Dispose() {
                         }
                         #endregion
                         #region IEnumerator implementation
                         public T Current {
                             get {
                                 if (m_pos == -1 || m_current == null)
                                     throw new InvalidOperationException();
                                 return m_current.GetAt(m_pos);
                             }
                         }
                         #endregion
                     }
                     public IEnumerator<T> GetEnumerator() {
                         return new Enumerator(m_first);
                     }
                     #endregion
                     #region IEnumerable implementation
                     IEnumerator IEnumerable.GetEnumerator() {
                         return GetEnumerator();
                     }
                     #endregion
                 }
             }

Implab/PromiseExtensions.cs +56 -3

             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(null,null,() => head.On(cleanup));
                         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 Combine(this ICollection<IPromise> that) {
+                    public static IPromise Bundle(this ICollection<IPromise> that) {
                         Safe.ArgumentNotNull(that, "that");
                         int count = that.Count;
+                        int errors = 0;
                         var medium = new Promise();
+                        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 => {
-                                    throw new Exception("The dependency promise is failed", error);
+                                    if (Interlocked.Increment(ref errors) == 1)
+                                        medium.Reject(
+                                            new Exception("The dependency promise is failed", error)
+                                        );
                                 },
                                 () => {
-                                    throw new OperationCanceledException("The dependency promise is cancelled");
+                                    if (Interlocked.Increment(ref errors) == 1)
+                                        medium.Reject(
+                                            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)
+                                        );
+                                },
+                                () => {
+                                    if (Interlocked.Increment(ref errors) == 1)
+                                        medium.Reject(
+                                            new Exception("The dependency promise is cancelled")
+                                        );
+                                }
+                            );
+                            i++;
+                        }
+                        return medium;
+                    }
                     #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, tcs.SetCanceled);
                         return tcs.Task;
                     }
                     #endif
                 }
             }

MonoPlay/Program.cs +2 -2

             using System;
             using Implab.Diagnostics;
             using Implab.Parallels;
             using Implab;
             using System.Collections.Generic;
             using System.Collections.Concurrent;
             namespace MonoPlay {
                 class MainClass {
                     public static void Main(string[] args) {
                         if (args == null)
                             throw new ArgumentNullException("args");
                         var q1 = new AsyncQueue<int>();
                         var q2 = new Queue<int>();
                         const int count = 10000000;
                         int res1 = 0, res2 = 0;
                         var t1 = Environment.TickCount;
                         AsyncPool.RunThread(
                             () => {
                                 for (var i = 0; i < count; i++)
                                     q1.Enqueue(1);
                                 Console.WriteLine("done writer #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 for (var i = 0; i < count; i++)
                                     q1.Enqueue(2);
                                 Console.WriteLine("done writer #2: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 int temp = 0;
                                 int i = 0;
                                 while (i < count)
                                     if (q1.TryDequeue(out temp)) {
                                         i++;
                                         res1 += temp;
                                     }
                                 Console.WriteLine("done reader #1: {0} ms", Environment.TickCount - t1);
                             },
                             () => {
                                 int temp = 0;
                                 int i = 0;
                                 while (i < count)
                                     if (q1.TryDequeue(out temp)) {
                                         i++;
                                         res2 += temp;
                                     }
                                 Console.WriteLine("done reader #2: {0} ms", Environment.TickCount - t1);
                             }
                         )
-                            .Combine()
+                            .Bundle()
                             .Join();
                         Console.WriteLine("done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}", Environment.TickCount - t1, res1, res2, res1 + res2, count*2);
                         var t2 = Environment.TickCount;
                         Console.WriteLine("MTQueue: {0} ms", t2 - t1);
                         t1 = Environment.TickCount;
                         for (var i = 0; i < count * 2; i++)
                             q2.Enqueue(i);
                         for (var i = 0; i < count * 2; i++)
                             q2.Dequeue();
                         t2 = Environment.TickCount;
                         Console.WriteLine("Queue: {0} ms", t2 - t1);
                         q2 = new Queue<int>();
                         t1 = Environment.TickCount;
                         AsyncPool.RunThread(
                             () => {
                                 for (var i = 0; i < count; i++)
                                     lock (q2)
                                         q2.Enqueue(i);
                             },
                             () => {
                                 for (var i = 0; i < count; i++)
                                     lock (q2)
                                         q2.Enqueue(i);
                             },
                             () => {
                                 for (int i = 0; i < count ;)
                                     lock (q2) {
                                         if (q2.Count == 0)
                                             continue;
                                         q2.Dequeue();
                                         i++;
                                     }
                             },
                             () => {
                                 for (int i = 0; i < count ;)
                                     lock (q2) {
                                         if (q2.Count == 0)
                                             continue;
                                         q2.Dequeue();
                                         i++;
                                     }
                             }
                         )
-                            .Combine()
+                            .Bundle()
                             .Join();
                         t2 = Environment.TickCount;
                         Console.WriteLine("Queue+Lock: {0} ms", t2 - t1);
                     }
                 }
             }

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