implab/ImplabNet Commit - r122:0c8685c8b56b

minor fixes and improvements of AsyncQueue, additional tests

cin -

r122:0c8685c8b56b v2

parent child

Context file:

r122:0c8685c8b56b

Collapse all files

Implab.Test/AsyncTests.cs +76 0

             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(() => {
                                     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(() => {
                                     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()
                             .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()
                             .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()
+                            .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/AsyncQueue.cs +19 -30

             using System.Threading;
             using System.Collections.Generic;
             using System;
             using System.Collections;
             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;
                         }
                         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 alloc;
-                            int allocSize;
+                            //int allocSize;
-                            // in case the batch size is larger than a free space in chunk
-                            // tailGap is used to over allocate the space in the chunk to
-                            // get exclusive permission on creation of the next one.
-                            int tailGap = 0;
-                            do {
-                                alloc = m_alloc;
-                                if (alloc > m_size) {
+                            var alloc = Interlocked.Add(ref m_alloc, length) - length;
-                                    // the chunk is full and someone already
+                            if (alloc > m_size) {
-                                    // creating the new one
+                                // the chunk is full and someone already
-                                    enqueued = 0; // nothing was added
+                                // creating the new one
-                                    extend = false; // the caller shouldn't try to extend the queue
+                                enqueued = 0; // nothing was added
-                                    return false; // nothing was added
+                                extend = false; // the caller shouldn't try to extend the queue
+                                return false; // nothing was added
+                            }
-                                allocSize = Math.Min(m_size - alloc, length);
+                            enqueued = Math.Min(m_size - alloc, length);
-                                if (allocSize < length) // the chunk doesn't have enough space to hold the whole batch
+                            extend = length > enqueued;
-                                    tailGap = 1; // overallocate space to get exclusive permission to extend queue
-                            } while(alloc != Interlocked.CompareExchange(ref m_alloc, alloc + allocSize + tailGap, alloc));
-                            extend = tailGap != 0;
-                            enqueued = allocSize;
-                            // if the chunk was already full then length > 0, allocSize = 0, tailGap = 1
+                            if (enqueued == 0)
-                            if (alloc == m_size)
                                 return false;
-                            Array.Copy(batch, offset, m_data, alloc, allocSize);
-                            while (alloc != Interlocked.CompareExchange(ref m_hi, alloc + allocSize, alloc)) {
+                            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;
+                                    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 data will be written.</param>
+                    /// <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 (next == null) {
                             // looks like this is the last chunk
                             if (first != Interlocked.CompareExchange(ref m_last, null, first)) {
                                 // race
                                 // maybe someone already recycled this chunk
                                 // or a new chunk has been appedned to the queue
                                 return; // give up
                             }
                             // 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;
                     }
                     #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
                 }
             }

MonoPlay/Program.cs +1 -1

             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()
                             .Join();
-                        Console.WriteLine("done: {0} ms, summ#1: {1}, summ#2: {2}, total: {3}", Environment.TickCount - t1, res1, res2, 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*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()
                             .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