implab/ImplabNet Commit - r125:f803565868a4

improved performance of promises

cin -

r125:f803565868a4 v2

parent child

Context file:

r125:f803565868a4

Collapse all files

Implab.Test/AsyncTests.cs +1 -1

             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
                                 .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(0,10,1)) {
+                        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 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/AbstractPromise.cs +80 -10

             using System;
             using Implab.Parallels;
             using System.Threading;
             using System.Reflection;
             namespace Implab {
                 public abstract class AbstractPromise<THandler> {
                     const int UNRESOLVED_SATE = 0;
                     const int TRANSITIONAL_STATE = 1;
                     const int SUCCEEDED_STATE = 2;
                     const int REJECTED_STATE = 3;
                     const int CANCELLED_STATE = 4;
+                    const int RESERVED_HANDLERS_COUNT = 4;
                     int m_state;
                     Exception m_error;
+                    int m_handlersCount;
-                    readonly AsyncQueue<THandler> m_handlers = new AsyncQueue<THandler>();
+                    readonly THandler[] m_handlers = new THandler[RESERVED_HANDLERS_COUNT];
+                    MTQueue<THandler> m_extraHandlers;
+                    int m_handlerPointer = -1;
+                    int m_handlersCommited;
                     #region state managment
                     bool BeginTransit() {
                         return UNRESOLVED_SATE == Interlocked.CompareExchange(ref m_state, TRANSITIONAL_STATE, UNRESOLVED_SATE);
                     }
                     void CompleteTransit(int state) {
                         if (TRANSITIONAL_STATE != Interlocked.CompareExchange(ref m_state, state, TRANSITIONAL_STATE))
                             throw new InvalidOperationException("Can't complete transition when the object isn't in the transitional state");
                     }
                     void WaitTransition() {
                         while (m_state == TRANSITIONAL_STATE) {
                             Thread.MemoryBarrier();
                         }
                     }
                     protected void BeginSetResult() {
                         if (!BeginTransit()) {
                             WaitTransition();
                             if (m_state != CANCELLED_STATE)
                                 throw new InvalidOperationException("The promise is already resolved");
                         }
                     }
                     protected void EndSetResult() {
                         CompleteTransit(SUCCEEDED_STATE);
                         OnSuccess();
                     }
                     /// <summary>
                     /// Выполняет обещание, сообщая об ошибке
                     /// </summary>
                     /// <remarks>
                     /// Поскольку обещание должно работать в многопточной среде, при его выполнении сразу несколько потоков
                     /// могу вернуть ошибку, при этом только первая будет использована в качестве результата, остальные
                     /// будут проигнорированы.
                     /// </remarks>
                     /// <param name="error">Исключение возникшее при выполнении операции</param>
                     /// <exception cref="InvalidOperationException">Данное обещание уже выполнено</exception>
                     protected void SetError(Exception error) {
                         if (BeginTransit()) {
                             m_error = error is PromiseTransientException ? error.InnerException : error;
                             CompleteTransit(REJECTED_STATE);
                             OnError();
                         } else {
                             WaitTransition();
                             if (m_state == SUCCEEDED_STATE)
                                 throw new InvalidOperationException("The promise is already resolved");
                         }
                     }
                     /// <summary>
                     /// Отменяет операцию, если это возможно.
                     /// </summary>
                     /// <remarks>Для определения была ли операция отменена следует использовать свойство <see cref="IsCancelled"/>.</remarks>
                     protected void SetCancelled() {
                         if (BeginTransit()) {
                             CompleteTransit(CANCELLED_STATE);
                             OnCancelled();
                         }
                     }
                     protected abstract void SignalSuccess(THandler handler);
                     protected abstract void SignalError(THandler handler, Exception error);
                     protected abstract void SignalCancelled(THandler handler);
                     void OnSuccess() {
+                        var hp = m_handlerPointer;
+                        var slot = hp +1 ;
+                        while (slot < m_handlersCommited) {
+                            if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
+                                SignalSuccess(m_handlers[slot]);
+                            }
+                            hp = m_handlerPointer;
+                            slot = hp +1 ;
+                        }
+                        if (m_extraHandlers != null) {
                             THandler handler;
-                        while (m_handlers.TryDequeue(out handler))
+                            while (m_extraHandlers.TryDequeue(out handler))
                                 SignalSuccess(handler);
                         }
+                    }
                     void OnError() {
+                        var hp = m_handlerPointer;
+                        var slot = hp +1 ;
+                        while (slot < m_handlersCommited) {
+                            if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
+                                SignalError(m_handlers[slot],m_error);
+                            }
+                            hp = m_handlerPointer;
+                            slot = hp +1 ;
+                        }
+                        if (m_extraHandlers != null) {
                             THandler handler;
-                        while (m_handlers.TryDequeue(out handler))
+                            while (m_extraHandlers.TryDequeue(out handler))
                                 SignalError(handler, m_error);
                         }
+                    }
                     void OnCancelled() {
+                        var hp = m_handlerPointer;
+                        var slot = hp +1 ;
+                        while (slot < m_handlersCommited) {
+                            if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) == hp) {
+                                SignalCancelled(m_handlers[slot]);
+                            }
+                            hp = m_handlerPointer;
+                            slot = hp +1 ;
+                        }
+                        if (m_extraHandlers != null) {
                             THandler handler;
-                        while (m_handlers.TryDequeue(out handler))
+                            while (m_extraHandlers.TryDequeue(out handler))
                                 SignalCancelled(handler);
                         }
+                    }
                     #endregion
                     protected abstract void Listen(PromiseEventType events, Action handler);
                     #region synchronization traits
                     protected void WaitResult(int timeout) {
                         if (!IsResolved) {
                             var lk = new object();
                             Listen(PromiseEventType.All, () => {
                                 lock(lk) {
                                     Monitor.Pulse(lk);
                                 }
                             });
                             lock (lk) {
                                 while(!IsResolved) {
                                     if(!Monitor.Wait(lk,timeout))
                                         throw new TimeoutException();
                                 }
                             }
                         }
                         switch (m_state) {
                             case SUCCEEDED_STATE:
                                 return;
                             case CANCELLED_STATE:
                                 throw new OperationCanceledException();
                             case REJECTED_STATE:
                                 throw new TargetInvocationException(m_error);
                             default:
                                 throw new ApplicationException(String.Format("Invalid promise state {0}", m_state));
                         }
                     }
                     #endregion
                     #region handlers managment
                     protected void AddHandler(THandler handler) {
-                        if (IsResolved) {
+                        if (m_state > 1) {
-                            InvokeHandler(handler);
-                        } else {
                             // the promise is in the resolved state, just invoke the handler
-                            m_handlers.Enqueue(handler);
+                            InvokeHandler(handler);
+                        } else {
+                            var slot = Interlocked.Increment(ref m_handlersCount) - 1;
+                            if (slot < RESERVED_HANDLERS_COUNT) {
+                                m_handlers[slot] = handler;
+                                while (slot != Interlocked.CompareExchange(ref m_handlersCommited, slot + 1, slot)) {
+                                }
+                                if (m_state > 1) {
+                                    do {
+                                        var hp = m_handlerPointer;
+                                        slot = hp + 1;
+                                        if (slot < m_handlersCommited) {
+                                            if (Interlocked.CompareExchange(ref m_handlerPointer, slot, hp) != hp)
+                                                continue;
+                                            InvokeHandler(m_handlers[slot]);
+                                        }
+                                        break;
+                                    } while(true);
+                                }
+                            } else {
+                                if (slot == RESERVED_HANDLERS_COUNT) {
+                                    m_extraHandlers = new MTQueue<THandler>();
+                                } else {
+                                    while (m_extraHandlers == null)
+                                        Thread.MemoryBarrier();
+                                }
-                            if (IsResolved && m_handlers.TryDequeue(out handler))
+                                m_extraHandlers.Enqueue(handler);
+                                if (m_state > 1 && m_extraHandlers.TryDequeue(out handler))
                                 // if the promise have been resolved while we was adding the handler to the queue
                                 // we can't guarantee that someone is still processing it
                                 // therefore we need to fetch a handler from the queue and execute it
                                 // note that fetched handler may be not the one that we have added
                                 // even we can fetch no handlers at all :)
                                 InvokeHandler(handler);
                             }
                         }
+                    }
                     protected void InvokeHandler(THandler handler) {
                         switch (m_state) {
                             case SUCCEEDED_STATE:
                                 SignalSuccess(handler);
                                 break;
                             case CANCELLED_STATE:
                                 SignalCancelled(handler);
                                 break;
                             case REJECTED_STATE:
                                 SignalError(handler, m_error);
                                 break;
                             default:
                                 throw new Exception(String.Format("Invalid promise state {0}", m_state));
                         }
                     }
                     #endregion
                     #region IPromise implementation
                     public void Join(int timeout) {
                         WaitResult(timeout);
                     }
                     public void Join() {
                         WaitResult(-1);
                     }
                     public bool IsResolved {
                         get {
                             Thread.MemoryBarrier();
                             return m_state > 1;
                         }
                     }
                     public bool IsCancelled {
                         get {
                             Thread.MemoryBarrier();
                             return m_state == CANCELLED_STATE;
                         }
                     }
                     #endregion
                     #region ICancellable implementation
                     public void Cancel() {
                         SetCancelled();
                     }
                     #endregion
                 }
             }

Implab/Parallels/AsyncQueue.cs +5 -11

             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);
+                                var chunk = new Chunk(DEFAULT_CHUNK_SIZE, 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),
+                                        Math.Max(size, DEFAULT_CHUNK_SIZE),
                                         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 {
                             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) {
                             if (first != Interlocked.CompareExchange(ref m_last, null, first)) {
                                 /*while (first.next == null)
                                     Thread.MemoryBarrier();*/
                                 // race
                                 // someone already updated the tail, restore the pointer to the queue head
                                 m_first = first;
                             }
                             // the tail is updated
                         }
                         // we need to update the head
                         //Interlocked.CompareExchange(ref m_first, next, first);
                         // if the head is already updated then give up
                         //return;
                     }
                     public void Clear() {
                         // start the new queue
-                        var chunk = new Chunk(m_chunkSize);
+                        var chunk = new Chunk(DEFAULT_CHUNK_SIZE);
                         do {
                             Thread.MemoryBarrier();
                             var first = m_first;
                             var last = m_last;
                             if (last == null) // nothing to clear
                                 return;
                             if (first == null || (first.next == null && first != last)) // inconcistency
                                 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
                             m_last = chunk;
                             return;
                         } while(true);
                     }
                     public T[] Drain() {
                         // start the new queue
-                        var chunk = new Chunk(m_chunkSize);
+                        var chunk = new Chunk(DEFAULT_CHUNK_SIZE);
                         do {
                             Thread.MemoryBarrier();
                             var first = m_first;
                             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, object last) {
                         var result = new List<T>();
-                        var buffer = new T[m_chunkSize];
+                        var buffer = new T[DEFAULT_CHUNK_SIZE];
                         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));
                             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/Parallels/DispatchPool.cs +1 -3

             using System;
             using System.Threading;
             namespace Implab.Parallels {
                 public abstract class DispatchPool<TUnit> : IDisposable {
                     readonly int m_minThreadsLimit;
                     readonly int m_maxThreadsLimit;
                     readonly int m_releaseTimeout = 1000; // the timeout while the working thread will wait for the new tasks before exit
                     int m_threads; // the current size of the pool
                     int m_maxRunningThreads; // the meximum reached size of the pool
                     int m_exit; // the pool is going to shutdown, all unused workers are released
                     readonly object m_signal = new object(); // used to pulse waiting threads
                     protected DispatchPool(int min, int max) {
                         if (min < 0)
                             throw new ArgumentOutOfRangeException("min");
                         if (max <= 0)
                             throw new ArgumentOutOfRangeException("max");
                         if (min > max)
                             min = max;
                         m_minThreadsLimit = min;
                         m_maxThreadsLimit = max;
                     }
                     protected DispatchPool(int threads)
                         : this(threads, threads) {
                     }
                     protected DispatchPool() {
-                        int maxThreads, maxCP;
-                        ThreadPool.GetMaxThreads(out maxThreads, out maxCP);
                         m_minThreadsLimit = 0;
-                        m_maxThreadsLimit = maxThreads;
+                        m_maxThreadsLimit = Environment.ProcessorCount;
                     }
                     protected void InitPool() {
                         for (int i = 0; i < m_minThreadsLimit; i++)
                             StartWorker();
                     }
                     public int PoolSize {
                         get {
                             Thread.MemoryBarrier();
                             return m_threads;
                         }
                     }
                     public int MaxRunningThreads {
                         get {
                             Thread.MemoryBarrier();
                             return m_maxRunningThreads;
                         }
                     }
                     protected bool IsDisposed {
                         get {
                             Thread.MemoryBarrier();
                             return m_exit == 1;
                         }
                     }
                     protected abstract bool TryDequeue(out TUnit unit);
                     bool Dequeue(out TUnit unit, int timeout) {
                         int ts = Environment.TickCount;
                         if (TryDequeue(out unit))
                             return true;
                         lock (m_signal) {
                             while (!TryDequeue(out unit) && m_exit == 0)
                                 if(!Monitor.Wait(m_signal, Math.Max(0, ts + timeout - Environment.TickCount))) {
                                     // timeout
                                     return false;
                                 }
                             // queue item or terminate
                             Monitor.Pulse(m_signal);
                             if (m_exit == 1)
                                 return false;
                         }
                         return true;
                     }
                     protected void SignalThread() {
                         lock (m_signal) {
                             Monitor.Pulse(m_signal);
                         }
                     }
                     #region thread slots traits
                     bool AllocateThreadSlot() {
                         int current;
                         // use spins to allocate slot for the new thread
                         do {
                             current = m_threads;
                             if (current >= m_maxThreadsLimit || m_exit == 1)
                                 // no more slots left or the pool has been disposed
                                 return false;
                         } while (current != Interlocked.CompareExchange(ref m_threads, current + 1, current));
                         UpdateMaxThreads(current + 1);
                         return true;
                     }
                     bool AllocateThreadSlot(int desired) {
                         if (desired - 1 != Interlocked.CompareExchange(ref m_threads, desired, desired - 1))
                             return false;
                         UpdateMaxThreads(desired);
                         return true;
                     }
                     bool ReleaseThreadSlot(out bool last) {
                         last = false;
                         int current;
                         // use spins to release slot for the new thread
                         Thread.MemoryBarrier();
                         do {
                             current = m_threads;
                             if (current <= m_minThreadsLimit && m_exit == 0)
                                 // the thread is reserved
                                 return false;
                         } while (current != Interlocked.CompareExchange(ref m_threads, current - 1, current));
                         last = (current == 1);
                         return true;
                     }
                     void UpdateMaxThreads(int count) {
                         int max;
                         do {
                             max = m_maxRunningThreads;
                             if (max >= count)
                                 break;
                         } while(max != Interlocked.CompareExchange(ref m_maxRunningThreads, count, max));
                     }
                     #endregion
                     protected bool StartWorker() {
                         if (AllocateThreadSlot()) {
                             // slot successfully allocated
                             var worker = new Thread(Worker);
                             worker.IsBackground = true;
                             worker.Start();
                             return true;
                         }
                         return false;
                     }
                     protected abstract void InvokeUnit(TUnit unit);
                     protected virtual void Worker() {
                         TUnit unit;
                         bool last;
                         do {
                             while (Dequeue(out unit, m_releaseTimeout)) {
                                 InvokeUnit(unit);
                             }
                             if(!ReleaseThreadSlot(out last))
                                 continue;
                             // queue may be not empty
                             if (last && TryDequeue(out unit)) {
                                 InvokeUnit(unit);
                                 if (AllocateThreadSlot(1))
                                     continue;
                                 // we can safely exit since pool is alive
                             }
                             break;
                         } while(true);
                     }
                     protected virtual void Dispose(bool disposing) {
                         if (disposing) {
                             if (0 == Interlocked.CompareExchange(ref m_exit, 1, 0)) { // implies memory barrier
                                 // wake sleeping threads
                                 SignalThread();
                                 GC.SuppressFinalize(this);
                             }
                         }
                     }
                     public void Dispose() {
                         Dispose(true);
                     }
                     ~DispatchPool() {
                         Dispose(false);
                     }
                 }
             }

MonoPlay/Program.cs +6 -92

@@ -1,120 +1,34
1	using System;	1	using System;
2	using Implab.Diagnostics;	2	using Implab.Diagnostics;
3	using Implab.Parallels;	3	using Implab.Parallels;
4	using Implab;	4	using Implab;
5	using System.Collections.Generic;	5	using System.Collections.Generic;
6	using System.Collections.Concurrent;	6	using System.Collections.Concurrent;
7		7
8	namespace MonoPlay {	8	namespace MonoPlay {
9	class MainClass {	9	class MainClass {
10	public static void Main(string[] args) {	10	public static void Main(string[] args) {
11	if (args == null)	11	if (args == null)
12	throw new ArgumentNullException("args");	12	throw new ArgumentNullException("args");
13		13
14	var q1 = new AsyncQueue<int>();
15	var q2 = new Queue<int>();
16
17	const int count = 10000000;	14	const int count = 10000000;
18		15
19	int res1 = 0, res2 = 0;
20	var t1 = Environment.TickCount;	16	var t1 = Environment.TickCount;
21		17
22	AsyncPool.RunThread(	18	for (int i = 0; i < count; i++) {
23	() => {	19	var p = new Promise<int>();
24	for (var i = 0; i < count; i++)
25	q1.Enqueue(1);
26	Console.WriteLine("done writer #1: {0} ms", Environment.TickCount - t1);
27	},
28	() => {
29	for (var i = 0; i < count; i++)
30	q1.Enqueue(2);
31	Console.WriteLine("done writer #2: {0} ms", Environment.TickCount - t1);
32	},
33	() => {
34	int temp = 0;
35	int i = 0;
36	while (i < count)
37	if (q1.TryDequeue(out temp)) {
38	i++;
39	res1 += temp;
40	}
41	Console.WriteLine("done reader #1: {0} ms", Environment.TickCount - t1);
42	},
43	() => {
44	int temp = 0;
45	int i = 0;
46	while (i < count)
47	if (q1.TryDequeue(out temp)) {
48	i++;
49	res2 += temp;
50	}
51	Console.WriteLine("done reader #2: {0} ms", Environment.TickCount - t1);
52	}
53	)
54	.Bundle()
55	.Join();
56
57	Console.WriteLine("done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}, count: {4}", Environment.TickCount - t1, res1, res2, res1 + res2, count*2);
58
59	var t2 = Environment.TickCount;
60	Console.WriteLine("MTQueue: {0} ms", t2 - t1);
61
62	t1 = Environment.TickCount;
63		20
64	for (var i = 0; i < count * 2; i++)	21	p.On(x => {}).On(x => {});
65	q2.Enqueue(i);
66
67	for (var i = 0; i < count * 2; i++)
68	q2.Dequeue();
69
70	t2 = Environment.TickCount;
71	Console.WriteLine("Queue: {0} ms", t2 - t1);
72
73	q2 = new Queue<int>();
74
75	t1 = Environment.TickCount;
76
77		22
78	AsyncPool.RunThread(	23	p.Resolve(i);
79	() => {
80	for (var i = 0; i < count; i++)
81	lock (q2)
82	q2.Enqueue(i);
83	},
84	() => {
85	for (var i = 0; i < count; i++)
86	lock (q2)
87	q2.Enqueue(i);
88	},
89	() => {
90	for (int i = 0; i < count ;)
91	lock (q2) {
92	if (q2.Count == 0)
93	continue;
94	q2.Dequeue();
95	i++;
96	}
97
98	},
99	() => {
100	for (int i = 0; i < count ;)
101	lock (q2) {
102	if (q2.Count == 0)
103	continue;
104	q2.Dequeue();
105	i++;
106	}
107		24
108	}	25	}
109	)
110	.Bundle()
111	.Join();
112		26
113		27
114		28
115	t2 = Environment.TickCount;	29	var t2 = Environment.TickCount;
116	Console.WriteLine("Queue+Lock: {0} ms", t2 - t1);	30	Console.WriteLine("done: {0} ms, {1:.00} Mb, {2} GC", t2 - t1, GC.GetTotalMemory(false) / (1024*1024), GC.CollectionCount(0) );
117		31
118	}	32	}
119	}	33	}
120	}	34	}

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