org.elasticsearch.common.util.concurrent

Class ThreadBarrier


  • public class ThreadBarrier
extends CyclicBarrier
    A synchronization aid that allows a set of threads to all wait for each other to reach a common barrier point. Barriers are useful in programs involving a fixed sized party of threads that must occasionally wait for each other. ThreadBarrier adds a cause to BrokenBarrierException thrown by a CyclicBarrier.reset() operation defined by CyclicBarrier.

    Sample usage:
  • Barrier as a synchronization and Exception handling aid
  • Barrier as a trigger for elapsed notification events 

    •     class MyTestClass implements RemoteEventListener
    {
      final ThreadBarrier barrier;

      class Worker implements Runnable
        {
          public void run()
            {
              barrier.await();    //wait for all threads to reach run
              try
                {
                  prepare();
                  barrier.await();    //wait for all threads to prepare
                  process();
                  barrier.await();    //wait for all threads to process
                }
              catch(Throwable t){
                  log("Worker thread caught exception", t);
                  barrier.reset(t);
                }
            }
        }

      public void testThreads() {
          barrier = new ThreadBarrier(N_THREADS + 1);
          for (int i = 0; i < N; ++i)
           new Thread(new Worker()).start();

          try{
              barrier.await();    //wait for all threads to reach run
              barrier.await();    //wait for all threads to prepare
              barrier.await();    //wait for all threads to process
            }
          catch(BrokenBarrierException bbe) {
              Assert.fail(bbe);
            }
       }

      int actualNotificationCount = 0;
      public synchronized void notify (RemoteEvent event) {
          try{
              actualNotificationCount++;
              if (actualNotificationCount == EXPECTED_COUNT)
                  barrier.await();    //signal when all notifications arrive

               // too many notifications?
               Assert.assertFalse("Exceeded notification count",
                                          actualNotificationCount > EXPECTED_COUNT);
            }
          catch(Throwable t) {
              log("Worker thread caught exception", t);
              barrier.reset(t);
            }
        }

      public void testNotify() {
          barrier = new ThreadBarrier(N_LISTENERS + 1);
          registerNotification();
          triggerNotifications();

          //wait until either all notifications arrive, or
          //until a MAX_TIMEOUT is reached.
          barrier.await(MAX_TIMEOUT);

          //check if all notifications were accounted for or timed-out
          Assert.assertEquals("Notification count",
                                      EXPECTED_COUNT, actualNotificationCount);

          //inspect that the barrier isn't broken
          barrier.inspect(); //throws BrokenBarrierException if broken
        }
    }