protected override long RunDisruptorPass()
        {
            Setup();
            var latch = new CountdownEvent(NUM_EVENT_PROCESSORS);
            var listTh = new List<Thread>();
            for (int i = 0; i < 3; i++)
            {
                handlers[i].Reset(latch, -1 + ITERATIONS);
            }

            disruptor.Start();
            var start = System.Diagnostics.Stopwatch.StartNew();

            for (long i = 0; i < ITERATIONS; i++)
            {
                long sequence = ringBuffer.Next();
                ringBuffer[sequence].Value = i;
                ringBuffer.Publish(sequence);
            }
            latch.Wait();
            long opsPerSecond = (ITERATIONS * 1000L) / start.ElapsedMilliseconds;
            for (int i = 0; i < NUM_EVENT_PROCESSORS; i++)
            {
                Assert.AreEqual(results[i], handlers[i].Value);
            }
            disruptor.Shutdown();
            return opsPerSecond;
        }

 public ConcurrentRunner(int maxThread, int loopEach)
 {
     this.maxThread = maxThread;
     this.loopEach = loopEach;
     this.semaphore = new SemaphoreSlim(0, maxThread);
     this.countdown = new CountdownEvent(maxThread);
 }

        protected PerformanceRecord ExecuteReadWithParallel(string operation, IEnumerable<uint> ids, int numberOfThreads, Func<long> readAction)
        {
            var countdownEvent = new CountdownEvent(numberOfThreads);

            var sw = Stopwatch.StartNew();
            var bytes = new long[numberOfThreads];
            for (int i = 0; i < numberOfThreads; i++)
            {
                var c = i;
                ThreadPool.QueueUserWorkItem(
                    state =>
                    {
                        bytes[c] = readAction();

                        countdownEvent.Signal();
                    });
            }

            countdownEvent.Wait();
            sw.Stop();

            return new PerformanceRecord
            {
                Bytes = bytes.Sum(),
                Operation = operation,
                Time = DateTime.Now,
                Duration = sw.ElapsedMilliseconds,
                ProcessedItems = ids.Count() * numberOfThreads
            };
        }

        public MainWindow()
        {
            InitializeComponent();
            Thread.CurrentThread.Name = "Main Thread";

            CountdownEvent folderEndByXML = new CountdownEvent(1);
            CountdownEvent folderEndByTree = new CountdownEvent(1);
            CountdownEvent fileEndByXML = new CountdownEvent(1);
            CountdownEvent fileEndByTree = new CountdownEvent(1);

            m_parser = new Parser(m_progressToken, folderEndByTree, folderEndByXML, fileEndByTree, fileEndByXML);
            m_treeFiller = new TreeFiller(m_treeView, m_progressToken, folderEndByTree, fileEndByTree);
            m_xmlFiller = new XMLFiller(m_progressToken, folderEndByXML, fileEndByXML);

            m_folderName.DataContext = m_parser;
            m_xmlFileName.DataContext = m_xmlFiller;

            m_parser.ItemGrabbed += m_xmlFiller.ItemGrabbedHandler;
            m_parser.ItemGrabbed += m_treeFiller.ItemGrabbedHandler;
            m_parser.FolderStarted += m_xmlFiller.FolderStartedHandler;
            m_parser.FolderStarted += m_treeFiller.FolderStartedHandler;
            m_parser.FolderFinished += m_xmlFiller.FolderFinishedHandler;
            m_parser.FolderFinished += m_treeFiller.FolderFinishedHandler;

            m_parser.ParserFinishEvent += this.ParserFinishEventHandler;

            m_xmlFiller.ExceptionOccuredEvent += this.ExceptionOccuredHandler;
            m_treeFiller.ExceptionOccuredEvent += this.ExceptionOccuredHandler;
            m_parser.ExceptionOccuredEvent += this.ExceptionOccuredHandler;
        }

        public IDisposable Start()
        {
            _monitor.Start();
            CountdownEvent pending = new CountdownEvent(_args.ConnectionLimit);
            var interval = Observable.Interval(TimeSpan.FromSeconds(1))
                .TakeWhile(_ => pending.CurrentCount > 0)
                .Subscribe(async _ =>
                {
                    var parallelCount = Math.Min(pending.CurrentCount, 10);

                    Task[] tasks = new Task[parallelCount];
                    for (int i = 0; i < parallelCount; i++)
                    {
                        tasks[i] = Task.Run(() => Connect(pending));
                    }

                    Task.WaitAll(tasks);
                },
                ex =>
                {
                    Console.WriteLine(ex.Message);
                    Environment.Exit(1);
                });

            pending.Wait();

            Observable.Interval(TimeSpan.FromSeconds(1)).Subscribe(this.SendMessage);

            return null;
        }

            public void should_reconnect_within_5_seconds()
            {
                const int total = 100;
                var countdown = new CountdownEvent(total);

                IBus producer = this.StartBus("producer", cfg => cfg.Route("boo"));

                IBus consumer = this.StartBus(
                    "consumer",
                    cfg => cfg.On<BooMessage>("boo").
                               ReactWith(
                                   (m, ctx) =>
                                       {
                                           Console.WriteLine("Received {0}.", m.Num);
                                           countdown.Signal();
                                       }));

                int count = total;
                while (count -- > 0)
                {
                    producer.Emit("boo", new BooMessage(count));
                    Console.WriteLine("Sent {0}.", count);
                    Thread.Sleep(1.Seconds());
                }

                countdown.Wait();
            }

		public void Constructor_Zero ()
		{
			var ce = new CountdownEvent (0);
			Assert.IsTrue (ce.IsSet, "#1");
			Assert.AreEqual (0, ce.InitialCount, "#2");
			Assert.IsTrue (ce.Wait (0), "#3");
		}

        protected override long RunDisruptorPass()
        {
            CountdownEvent latch = new CountdownEvent(1);
            long expectedCount = batchEventProcessor.Sequence.Value + ITERATIONS;
            handler.Reset(latch, ITERATIONS);
            Task.Factory.StartNew(() => batchEventProcessor.Run());
            Stopwatch start = Stopwatch.StartNew();

            RingBuffer<long[]> rb = ringBuffer;

            for (long i = 0; i < ITERATIONS; i++)
            {
                long next = rb.Next();
                long[] @event = rb.Get(next);
                for (int j = 0; j < @event.Length; j++)
                {
                    @event[j] = i;
                }
                rb.Publish(next);
            }

            latch.Wait();
            long opsPerSecond = (ITERATIONS * ARRAY_SIZE * 1000L) / (start.ElapsedMilliseconds);
            waitForEventProcessorSequence(expectedCount);
            batchEventProcessor.Halt();

            PerfTestUtil.failIf(0, handler.Value);

            return opsPerSecond;
        }

        public void Should_be_able_to_subscribe_as_exlusive()
        {
            var countdownEvent = new CountdownEvent(10);
            var firstCount = 0;
            var secondCount = 0;

            bus.Subscribe<MyMessage>("test", message =>
                {
                    countdownEvent.Signal();
                    Interlocked.Increment(ref firstCount);
                    Console.WriteLine("[1] " + message.Text);
                }, x => x.AsExclusive());
            bus.Subscribe<MyMessage>("test", message =>
                {
                    countdownEvent.Signal();
                    Interlocked.Increment(ref secondCount);
                    Console.WriteLine("[2] " + message.Text);
                }, x => x.AsExclusive());

            for (var i = 0; i < 10; ++i)
                bus.Publish(new MyMessage
                    {
                        Text = "Exclusive " + i
                    });
            countdownEvent.Wait(10 * 1000);
            Assert.IsTrue(firstCount == 10 && secondCount == 0 || firstCount == 0 && secondCount == 10);
            Console.WriteLine("Stopped consuming");
        }

        public void TestTimerStartAutoReset()
        {
            CountdownEvent cde = new CountdownEvent(1);
            int result = 0;
            _timer = new TestTimer(1);

            // Test defaults.
            Assert.Equal(1, _timer.Interval);
            Assert.True(_timer.AutoReset);

            _timer.AutoReset = false;
            _timer.Elapsed += (sender, e) => { result = ++result; cde.Signal(); };
            _timer.Start();

            Assert.True(_timer.Enabled);
            cde.Wait();

            // Only elapsed once.
            Assert.Equal(1, result);

            cde = new CountdownEvent(10);
            _timer.AutoReset = true;

            cde.Wait();
            cde.Dispose();

            _timer.Stop();
            // Atleast elapsed 10 times.
            Assert.True(result >= 10);
        }

        private static void Run(string[] args)
        {
            Environment.SetEnvironmentVariable("hazelcast.logging.level", "info");
            Environment.SetEnvironmentVariable("hazelcast.logging.type", "console");

            var config = new ClientConfig();
            config.GetNetworkConfig().AddAddress("127.0.0.1");
            var client = HazelcastClient.NewHazelcastClient(config);

            var list = client.GetList<string>("collection-listener-example");
            var cdown = new CountdownEvent(3);
            list.AddItemListener(new ItemListener<string>
            {
                OnItemAdded = e =>
                {
                    Console.WriteLine("Item added: " + e.GetItem());
                    cdown.Signal();
                },
                OnItemRemoved = e =>
                {
                    Console.WriteLine("Item removed: " + e.GetItem());
                    cdown.Signal();
                }
            }, true);

            list.Add("item1");
            list.Add("item2");
            list.Remove("item1");

            cdown.Wait();
            list.Destroy();
            client.Shutdown();
        }

        static void M1()
        {
            var sameLocalVariable = 123;
            var cdevent = new CountdownEvent(2);

            if (Fork.CloneThread())
            {
                lock (_sync)
                {
                    Console.ReadKey();
                    Console.WriteLine("in forked thread: {0}, tid: {1} ", sameLocalVariable, Thread.CurrentThread.ManagedThreadId);
                    cdevent.Signal();
                }
            }
            else
            {
                lock (_sync)
                {
                    Console.ReadKey();
                    Console.WriteLine("in parent thread: {0}, tid: {1} ", sameLocalVariable, Thread.CurrentThread.ManagedThreadId);
                    cdevent.Signal();
                }
            }

            cdevent.Wait();
        }

        public void be_able_to_subscribe_to_non_existing_stream_and_then_catch_event()
        {
            const string stream = "be_able_to_subscribe_to_non_existing_stream_and_then_catch_event";
            using (var store = BuildConnection(_node))
            {
                store.ConnectAsync().Wait();
                var appeared = new CountdownEvent(1);
                var dropped = new CountdownEvent(1);

                var subscription = store.SubscribeToStreamFrom(stream,
                                                               null,
                                                               CatchUpSubscriptionSettings.Default,
                                                               (_, x) => appeared.Signal(),
                                                               _ => Log.Info("Live processing started."),
                                                               (_, __, ___) => dropped.Signal());

                store.AppendToStreamAsync(stream, ExpectedVersion.EmptyStream, TestEvent.NewTestEvent()).Wait();

                if (!appeared.Wait(Timeout))
                {
                    Assert.IsFalse(dropped.Wait(0), "Subscription was dropped prematurely.");
                    Assert.Fail("Appeared countdown event timed out.");
                }

                Assert.IsFalse(dropped.Wait(0));
                subscription.Stop(Timeout);
                Assert.IsTrue(dropped.Wait(Timeout));
            }
        }

        public void be_able_to_subscribe_to_non_existing_stream()
        {
            const string stream = "be_able_to_subscribe_to_non_existing_stream";
            using (var store = BuildConnection(_node))
            {
                store.ConnectAsync().Wait();
                var appeared = new ManualResetEventSlim(false);
                var dropped = new CountdownEvent(1);

                var subscription = store.SubscribeToStreamFrom(stream,
                                                               null,
                                                               CatchUpSubscriptionSettings.Default,
                                                               (_, x) => appeared.Set(),
                                                               _ => Log.Info("Live processing started."),
                                                               (_, __, ___) => dropped.Signal());

                Thread.Sleep(100); // give time for first pull phase
                store.SubscribeToStreamAsync(stream, false, (s, x) => { }, (s, r, e) => { }).Wait();
                Thread.Sleep(100);
                Assert.IsFalse(appeared.Wait(0), "Some event appeared.");
                Assert.IsFalse(dropped.Wait(0), "Subscription was dropped prematurely.");
                subscription.Stop(Timeout);
                Assert.IsTrue(dropped.Wait(Timeout));
            }
        }

        public void HydratingFromMultipleThreads_IsSafe()
        {
            var numberOfHydrations = 0;
            var cache = new CachedValue<int>(() => ++numberOfHydrations);

            Assert.That(cache.Value, Is.EqualTo(1));

            using (var countdownEvent = new CountdownEvent(2))
            {
                Action threadAction = () =>
                {
                    cache.Invalidate();
                    countdownEvent.Signal();
                    countdownEvent.Wait();

                    Assert.That(cache.Value, Is.EqualTo(2));
                };

                var t1 = threadAction.BeginInvoke(threadAction.EndInvoke, null);
                var t2 = threadAction.BeginInvoke(threadAction.EndInvoke, null);
                WaitHandle.WaitAll(new[]{t1.AsyncWaitHandle, t2.AsyncWaitHandle});
            }

            Assert.That(numberOfHydrations, Is.EqualTo(2));
        }

        public void TestFindAllEntities()
        {
            RunAndAwait( () =>
            {
              var entities = new List<FindAllEntityAsync>();
              var latch = new CountdownEvent( 10 );
              for( int i = 0; i < 10; i++ )
              {
            var findAllEntity = new FindAllEntityAsync {Name = "bot_#" + i, Age = 20 + i};
            Backendless.Persistence.Save( findAllEntity, new AsyncCallback<FindAllEntityAsync>( response =>
              {
                entities.Add( findAllEntity );
                latch.Signal();
              }, fault =>
                {
                  for( int j = 0; j < latch.CurrentCount; j++ )
                    latch.Signal();

                  FailCountDownWith( fault );
                } ) );
              }
              latch.Wait();

              Backendless.Persistence.Of<FindAllEntityAsync>()
                     .Find( new ResponseCallback<BackendlessCollection<FindAllEntityAsync>>( this )
                       {
                         ResponseHandler =
                           backendlessCollection => AssertArgumentAndResultCollections( entities, backendlessCollection )
                       } );
            } );
        }

        static void Main(string[] args)
        {
            // initialize the semaphores
            semA = new SemaphoreSlim(2);
            semB = new SemaphoreSlim(2);

            // define the number of tasks we will use
            int taskCount = 10;

            // initialize the barrier
            cdEvent = new CountdownEvent(taskCount);

            Task[] tasks = new Task[10];
            for (int i = 0; i < taskCount; i++) {
                tasks[i] = Task.Factory.StartNew((stateObject) => {
                    InitialMethod((int)stateObject);
                    Console.WriteLine("Task {0} completed", Task.CurrentId);
                }, i);
            }

            // wait for all of the tasks to have reached a terminal method
            cdEvent.Wait();

            // throw an exception to force the debugger to break
            throw new Exception();
        }

        public void Should_catch_EndOfStreamException()
        {
            // Arrange
            var count = new CountdownEvent(2);
            var channel = Substitute.For<IModel>();
            var watcher = Substitute.For<IRabbitWatcher>();
            var queue = Substitute.For<IInMemoryPriorityQueue<GenericPriorityMessage<BasicDeliverEventArgs>>>();
            queue.When(x => x.Dequeue()).Do(callInfo => {
                count.Signal();
                throw new EndOfStreamException();                
            });
            var consumer = new PriorityBurrowConsumer(channel, Substitute.For<IMessageHandler>(), watcher, true, 2);

            var sub = Substitute.For<CompositeSubscription>();
            sub.AddSubscription(new Subscription(channel) { ConsumerTag = "Burrow" });
            consumer.Init(queue, sub, 1, Guid.NewGuid().ToString());
            

            // Action
            consumer.Ready();
            count.Wait();

            // Assert
            consumer.Dispose();
        }

        public void Should_execute_less_than_4_seconds()
        {
            ThreadPool.SetMinThreads(NumThreads, NumThreads);

            Console.WriteLine("Burst test started");

            _mre.Reset();
            _countdown = new CountdownEvent(NumThreads);

            for (var i = 0; i < NumThreads; i++)
            {
                new Thread(OneThreadExecution) { Name = "Thread " + i }.Start();
            }

            _countdown.Wait();
            var dateTime = DateTime.Now;
            _countdown = new CountdownEvent(NumThreads);
            _mre.Set();
            _countdown.Wait();
            var timeSpan = DateTime.Now - dateTime;

            Console.WriteLine("Test finished");
            Console.WriteLine("Executed at {0}.{1:0}s.", timeSpan.Seconds, timeSpan.Milliseconds / 100);

            if (timeSpan.Seconds > 5)
            {
                Assert.Ignore("This test should't take more than to 4 seconds to run");
            }
        }

        public void SocketConnect()
        {
            const int port = 5050;
            const int clientCount = 10;
            var listener = SocketEvents.GetTcpStreamSockets(port);
            var countdown = new CountdownEvent(clientCount);
            try
            {
                var tcs = new TaskCompletionSource<object>();
                int count = 0;
                listener.Subscribe(s =>
                {
                    count++;
                    countdown.Signal();
                    s.Close();
                },
                tcs.SetException,
                () => tcs.TrySetResult(null));

                for (int i = 0; i < clientCount; i++)
                {
                    SocketTestUtility.Connect(port);
                }

                countdown.WaitEx();
                Assert.IsTrue(count == clientCount);
            }
            finally
            {
                listener.Dispose();
            }
        }