Java 类java.util.concurrent.CyclicBarrier 实例源码

/**
 * 版本3.0.9以下存在多线程初始化问题，这个类作为一个样例
 */
public static void main(String[] args) throws InterruptedException, BrokenBarrierException {
    System.out.println(Arrays.asList(splitWord));
    for (int j = 0; j < 1000; j++) {
        CyclicBarrier barrier = new CyclicBarrier(11, null);
        for (int i = 0; i < 10; i++) {
            Thread t = new Thread(new Worker(barrier), "t" + i);
            t.start();
        }
        Thread.sleep(500);
        barrier.await();
        while (barrier.getNumberWaiting() > 0) {
            Thread.sleep(1000);
        }
        Thread.sleep(1000);
        System.out.println(Arrays.asList(splitWord));
    }
}

/**
 * A reset before threads enter barrier does not throw
 * BrokenBarrierException
 */
public void testReset_NoBrokenBarrier() throws Exception {
    final CyclicBarrier c = new CyclicBarrier(3);
    c.reset();

    Thread t1 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws Exception {
            c.await();
        }});
    Thread t2 = newStartedThread(new CheckedRunnable() {
        public void realRun() throws Exception {
            c.await();
        }});

    c.await();
    awaitTermination(t1);
    awaitTermination(t2);
}

@Test(timeout=20000)
public void testCommitJobFailsJob() throws Exception {
  Configuration conf = new Configuration();
  conf.set(MRJobConfig.MR_AM_STAGING_DIR, stagingDir);
  AsyncDispatcher dispatcher = new AsyncDispatcher();
  dispatcher.init(conf);
  dispatcher.start();
  CyclicBarrier syncBarrier = new CyclicBarrier(2);
  OutputCommitter committer = new TestingOutputCommitter(syncBarrier, false);
  CommitterEventHandler commitHandler =
      createCommitterEventHandler(dispatcher, committer);
  commitHandler.init(conf);
  commitHandler.start();

  JobImpl job = createRunningStubbedJob(conf, dispatcher, 2, null);
  completeJobTasks(job);
  assertJobState(job, JobStateInternal.COMMITTING);

  // let the committer fail and verify the job fails
  syncBarrier.await();
  assertJobState(job, JobStateInternal.FAILED);
  dispatcher.stop();
  commitHandler.stop();
}

public static void main(String[] args) throws InterruptedException {
        final int N=10;
        Thread[] allSoldier=new Thread[N];
        boolean flag=false;
        CyclicBarrier cyclic=new CyclicBarrier(N,new BarrierRun(flag,N));

        System.out.println("集合队伍！");
        for(int i=0;i<N;++i){
            System.out.println("士兵 "+i+"报道！");
            allSoldier[i]=new Thread(new Soldier(cyclic,"士兵"+i));
            allSoldier[i].start();
//            if(i==5) {这个中断会引起一个interrupt()和n个BrokenBarrierException异常。意思是栅栏破坏掉了，
// 线程永远无法完成栅栏
//                allSoldier[1].interrupt();
//            }
        }
    }

/**
 * Blocks the named executor by getting its only thread running a task blocked on a CyclicBarrier and fills the queue with a noop task.
 * So requests to use this queue should get {@link EsRejectedExecutionException}s.
 */
private CyclicBarrier blockExecutor(String name) throws Exception {
    ThreadPool threadPool = getInstanceFromNode(ThreadPool.class);
    CyclicBarrier barrier = new CyclicBarrier(2);
    logger.info("Blocking the [{}] executor", name);
    threadPool.executor(name).execute(() -> {
        try {
            threadPool.executor(name).execute(() -> {});
            barrier.await();
            logger.info("Blocked the [{}] executor", name);
            barrier.await();
            logger.info("Unblocking the [{}] executor", name);
        } catch (Exception e) {
            throw new RuntimeException(e);
        }
    });
    barrier.await();
    blockedExecutors.add(barrier);
    return barrier;
}

public void testCustomScheduler_deadlock() throws InterruptedException, BrokenBarrierException {
  final CyclicBarrier inGetNextSchedule = new CyclicBarrier(2);
  // This will flakily deadlock, so run it multiple times to increase the flake likelihood
  for (int i = 0; i < 1000; i++) {
    Service service = new AbstractScheduledService() {
      @Override protected void runOneIteration() {}
      @Override protected Scheduler scheduler() {
        return new CustomScheduler() {
          @Override protected Schedule getNextSchedule() throws Exception {
            if (state() != State.STARTING) {
              inGetNextSchedule.await();
              Thread.yield();
              throw new RuntimeException("boom");
            }
            return new Schedule(0, TimeUnit.NANOSECONDS);
          }
        };
      }
    };
    service.startAsync().awaitRunning();
    inGetNextSchedule.await();
    service.stopAsync();
  }
}

static void test(int i, int nkeys, String[] key, Class mapClass) throws Exception {
  System.out.print("Threads: " + i + "\t:");
  Map map = (Map) mapClass.newInstance();
  // Uncomment to start with a non-empty table
  // for (int j = 0; j < nkeys; j += 4) // start 1/4 occupied
  // map.put(key[j], key[j]);
  LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
  CyclicBarrier barrier = new CyclicBarrier(i + 1, timer);
  for (int t = 0; t < i; ++t)
    pool.execute(new Runner(t, map, key, barrier));
  barrier.await();
  barrier.await();
  long time = timer.getTime();
  long tpo = time / (i * (long) nops);
  System.out.print(LoopHelpers.rightJustify(tpo) + " ns per op");
  double secs = (double) (time) / 1000000000.0;
  System.out.println("\t " + secs + "s run time");
  map.clear();
}

private SomeFeed(int threadCount, boolean barriered) {
        this.threadCount = threadCount;
        this.barriered = barriered;
        if (barriered) {
            barrier = new CyclicBarrier(threadCount, System.out::println);
        }
        launchPublishers();
        SomeListener tickOutputter = new SomeListener() {
            @Override
            public void priceTick(PriceTick event) {
                System.out.println(".");
            }

            @Override
            public void error(Throwable throwable) {

            }
        };
//        register(tickOutputter);
    }

public SimpleProcessorBenchmarkClientRunnable(String targetIP, int targetPort, int clientNums, int rpcTimeout,
                                              CyclicBarrier barrier, CountDownLatch latch, long startTime,
                                              long endTime){

    this.targetIP = targetIP;
    this.targetPort = targetPort;
    this.clientNums = clientNums;
    this.rpcTimeout = rpcTimeout;
    this.barrier = barrier;
    this.latch = latch;
    this.startTime = startTime;
    this.endTime = endTime;
    maxRange = (Integer.parseInt(String.valueOf((endTime - startTime))) / 1000) + 1;
    errorTPS = new long[maxRange];
    errorResponseTimes = new long[maxRange];
    tps = new long[maxRange];
    responseTimes = new long[maxRange];
    // init
    for (int i = 0; i < maxRange; i++) {
        errorTPS[i] = 0;
        errorResponseTimes[i] = 0;
        tps[i] = 0;
        responseTimes[i] = 0;
    }
}

/**
 * adds by multiple threads produce correct sum
 */
public void testAddAndSumMT() throws Throwable {
    final int incs = 1000000;
    final int nthreads = 4;
    final ExecutorService pool = Executors.newCachedThreadPool();
    LongAdder a = new LongAdder();
    CyclicBarrier barrier = new CyclicBarrier(nthreads + 1);
    for (int i = 0; i < nthreads; ++i)
        pool.execute(new AdderTask(a, barrier, incs));
    barrier.await();
    barrier.await();
    long total = (long)nthreads * incs;
    long sum = a.sum();
    assertEquals(sum, total);
    pool.shutdown();
}

@Test
public void eachThreadGetsDifferentGlobalTxId() throws Exception {
  CyclicBarrier barrier = new CyclicBarrier(2);

  Runnable runnable = exceptionalRunnable(() -> {
    String txId = UUID.randomUUID().toString();
    omegaContext.setGlobalTxId(txId);
    barrier.await();

    assertThat(omegaContext.globalTxId(), is(txId));
  });

  CompletableFuture<Void> future1 = CompletableFuture.runAsync(runnable);
  CompletableFuture<Void> future2 = CompletableFuture.runAsync(runnable);

  CompletableFuture.allOf(future1, future2).join();
}

@Test(timeout = 5000)
public void invokeEitherPostInterceptOrOnTimeoutConcurrently() throws Exception {
  List<Future<?>> futures = new LinkedList<>();

  for (int i = 0; i < runningCounts; i++) {
    TimeAwareInterceptor interceptor = new TimeAwareInterceptor(underlying);
    CyclicBarrier cyclicBarrier = new CyclicBarrier(2);


    futures.add(executorService.submit(() -> {
      waitForSignal(cyclicBarrier);
      interceptor.postIntercept(localTxId, signature);
    }));

    futures.add(executorService.submit(() -> {
      waitForSignal(cyclicBarrier);
      interceptor.onTimeout(localTxId, signature, timeoutException);
    }));
  }

  waitTillAllDone(futures);

  assertThat(postInterceptInvoked.get() + onTimeoutInvoked.get(), is(runningCounts));
}

@Test(timeout = 5000)
public void invokeEitherOnErrorOrOnTimeoutConcurrently() throws Exception {
  RuntimeException oops = new RuntimeException("oops");
  List<Future<?>> futures = new LinkedList<>();

  for (int i = 0; i < runningCounts; i++) {
    TimeAwareInterceptor interceptor = new TimeAwareInterceptor(underlying);
    CyclicBarrier cyclicBarrier = new CyclicBarrier(2);


    futures.add(executorService.submit(() -> {
      waitForSignal(cyclicBarrier);
      interceptor.onError(localTxId, signature, oops);
    }));

    futures.add(executorService.submit(() -> {
      waitForSignal(cyclicBarrier);
      interceptor.onTimeout(localTxId, signature, timeoutException);
    }));
  }

  waitTillAllDone(futures);

  assertThat(onErrorInvoked.get() + onTimeoutInvoked.get(), is(runningCounts));
}

public SimpleProcessorBenchmarkClientRunnable(String targetIP, int targetPort, int clientNums, int rpcTimeout,
                                              CyclicBarrier barrier, CountDownLatch latch, long startTime,
                                              long endTime){

    this.targetIP = targetIP;
    this.targetPort = targetPort;
    this.clientNums = clientNums;
    this.rpcTimeout = rpcTimeout;
    this.barrier = barrier;
    this.latch = latch;
    this.startTime = startTime;
    this.endTime = endTime;
    maxRange = (Integer.parseInt(String.valueOf((endTime - startTime))) / 1000) + 1;
    errorTPS = new long[maxRange];
    errorResponseTimes = new long[maxRange];
    tps = new long[maxRange];
    responseTimes = new long[maxRange];
    // init
    for (int i = 0; i < maxRange; i++) {
        errorTPS[i] = 0;
        errorResponseTimes[i] = 0;
        tps[i] = 0;
        responseTimes[i] = 0;
    }
}

public AbstractClientRunnable(String targetIP, int targetPort, int clientNums, int rpcTimeout,
                              CyclicBarrier barrier, CountDownLatch latch, long startTime, long endTime){

    this.barrier = barrier;
    this.latch = latch;
    this.startTime = startTime;
    this.endTime = endTime;
    serviceFactory.setTargetIP(targetIP);
    serviceFactory.setClientNums(clientNums);
    serviceFactory.setTargetPort(targetPort);
    serviceFactory.setConnectTimeout(rpcTimeout);
    maxRange = (Integer.parseInt(String.valueOf((endTime - startTime))) / 1000000) + 1;
    errorTPS = new long[maxRange];
    errorResponseTimes = new long[maxRange];
    tps = new long[maxRange];
    responseTimes = new long[maxRange];
    // init
    for (int i = 0; i < maxRange; i++) {
        errorTPS[i] = 0;
        errorResponseTimes[i] = 0;
        tps[i] = 0;
        responseTimes[i] = 0;
    }
}

public BidClientRunnable(String protocol, String serialization, String targetIP, int targetPort, int clientNums, int rpcTimeout,
                          CyclicBarrier barrier, CountDownLatch latch, long startTime,
                          long endTime){
    super(protocol, serialization, targetIP, targetPort, clientNums, rpcTimeout, barrier, latch, startTime, endTime);
    Impression imp = new Impression();
    imp.setBidFloor(1.1);
    imp.setId("abc");
    List<Impression> imps = new ArrayList<Impression>(1);
    imps.add(imp);
    request.setImpressions(imps);

    Geo geo = new Geo();
    geo.setCity("beijing");
    geo.setCountry("china");
    geo.setLat(100.1f);
    geo.setLon(100.1f);

    Device device = new Device();
    device.setMake("apple");
    device.setOs("ios");
    device.setVersion("7.0");
    device.setLang("zh_CN");
    device.setModel("iphone");
    device.setGeo(geo);
    request.setDevice(device);
}

public AbstractClientRunnable(String protocol, String serialization, String targetIP, int targetPort, int clientNums, int rpcTimeout,
                              CyclicBarrier barrier, CountDownLatch latch, long startTime, long endTime){

    this.barrier = barrier;
    this.latch = latch;
    this.startTime = startTime;
    this.endTime = endTime;
    serviceFactory.setProtocol(protocol);
    serviceFactory.setTargetIP(targetIP);
    serviceFactory.setClientNums(clientNums);
    serviceFactory.setTargetPort(targetPort);
    serviceFactory.setConnectTimeout(rpcTimeout);
    serviceFactory.setSerialization(serialization);
    maxRange = (Integer.parseInt(String.valueOf((endTime - startTime))) / 1000000) + 1;
    errorTPS = new long[maxRange];
    errorResponseTimes = new long[maxRange];
    tps = new long[maxRange];
    responseTimes = new long[maxRange];
    // init
    for (int i = 0; i < maxRange; i++) {
        errorTPS[i] = 0;
        errorResponseTimes[i] = 0;
        tps[i] = 0;
        responseTimes[i] = 0;
    }
}

public void await() {
    if(0 == delay) {
        log.info("No pause between retry");
        return;
    }
    final Timer wakeup = new Timer();
    final CyclicBarrier wait = new CyclicBarrier(2);
    // Schedule for immediate execution with an interval of 1s
    wakeup.scheduleAtFixedRate(new PauserTimerTask(wait), 0, 1000);
    try {
        // Wait for notify from wakeup timer
        wait.await();
    }
    catch(InterruptedException | BrokenBarrierException e) {
        log.error(e.getMessage(), e);
    }
}

synchronized void makeBarrier() {
        if (numPlayers < 1) {
            log.warning("cannot make barrier for " + numPlayers + " viewers - something is wrong");
            log.warning("disabling sychronized playback because probably multiple viewers are active but we are not playing set of sychronized files");
            outer.getToggleSyncEnabledAction().actionPerformed(null);
            // toggle all the viewers syncenabled menu item
//               JOptionPane.showMessageDialog(null,"Disabled sychronized playback because files are not part of sychronized set");
            return;
        }
        log.info("making barrier for " + this);
        barrier = new CyclicBarrier(numPlayers, new Runnable() {

            public void run() {
                // this is run after await synchronization; it updates the time to read events from each AEInputStream
//                        log.info(Thread.currentThread()+" resetting barrier");
                barrier.reset();
                setTime(getTime() + getTimesliceUs());
            }
        });
    }

public SimpleProcessorBenchmarkClientRunnable(String targetIP, int targetPort, int clientNums, int rpcTimeout,
                                              CyclicBarrier barrier, CountDownLatch latch, long startTime,
                                              long endTime){

    this.targetIP = targetIP;
    this.targetPort = targetPort;
    this.clientNums = clientNums;
    this.rpcTimeout = rpcTimeout;
    this.barrier = barrier;
    this.latch = latch;
    this.startTime = startTime;
    this.endTime = endTime;
    maxRange = (Integer.parseInt(String.valueOf((endTime - startTime))) / 1000) + 1;
    errorTPS = new long[maxRange];
    errorResponseTimes = new long[maxRange];
    tps = new long[maxRange];
    responseTimes = new long[maxRange];
    // init
    for (int i = 0; i < maxRange; i++) {
        errorTPS[i] = 0;
        errorResponseTimes[i] = 0;
        tps[i] = 0;
        responseTimes[i] = 0;
    }
}

static void oneRun(int nthreads, int iters) throws Exception {
    LoopHelpers.BarrierTimer timer = new LoopHelpers.BarrierTimer();
    CyclicBarrier barrier = new CyclicBarrier(nthreads + 1, timer);
    Exchanger<Int> l = null;
    Exchanger<Int> r = new Exchanger<>();
    for (int i = 0; i < nthreads; ++i) {
        pool.execute(new Stage(l, r, barrier, iters));
        l = r;
        r = (i+2 < nthreads) ? new Exchanger<Int>() : null;
    }
    barrier.await();
    barrier.await();
    long time = timer.getTime();
    if (print)
        System.out.println(LoopHelpers.rightJustify(time / (iters * nthreads + iters * (nthreads-2))) + " ns per transfer");
}

@Test
public void startsNewChildSpan() throws Exception {
  CyclicBarrier cyclicBarrier = new CyclicBarrier(nThreads);

  CompletableFuture<?>[] futures = new CompletableFuture[nThreads];
  for (int i = 0; i < nThreads; i++) {
    futures[i] = CompletableFuture.runAsync(() -> {
      Span currentSpan = tracing.tracer().newTrace().start();

      waitTillAllAreReady(cyclicBarrier);

      try (SpanInScope spanInScope = tracing.tracer().withSpanInScope(currentSpan)) {
        assertThat(tracingAdviser.invoke(spanName, path, supplier), is(expected));
      } catch (Throwable throwable) {
        fail(throwable.getMessage());
      } finally {
        currentSpan.finish();
      }
    }, Executors.newFixedThreadPool(nThreads));
  }

  CompletableFuture.allOf(futures).join();

  assertThat(traces.size(), is(nThreads));

  for (Queue<zipkin2.Span> queue : traces.values()) {
    zipkin2.Span child = queue.poll();
    assertThat(child.name(), is(spanName));

    zipkin2.Span parent = queue.poll();
    assertThat(child.parentId(), is(parent.id()));
    assertThat(child.traceId(), is(parent.traceId()));
    assertThat(tracedValues(child), contains(this.getClass().getCanonicalName()));
  }
}

public void testRebuildNoDeadlock() throws Exception {
    CyclicBarrier barrier= new CyclicBarrier(2);

    final Task1 task1 = new Task1();
    Thread t1 = new Thread(task1, "Thread 1");
    final Task2 task2 = new Task2(barrier);
    Thread t2 = new Thread(task2, "Thread 2");
    handler.setBarrier(barrier);
    t1.start();
    t2.start();
    t1.join(60000);
    t2.join(60000); // wait max 1 min for the test to finish
    assertTrue(task1.done);
    assertTrue(task2.done);
}

private static Awaiter awaiter(final CyclicBarrier barrier,
                               final long millis) {
    return new Awaiter() { public void run() {
        toTheStartingGate();

        try { barrier.await(millis, MILLISECONDS); }
        catch (Throwable result) { result(result); }}};
}

public static void main(String[] args) {
    int num = 10;
    CyclicBarrier barrier = new CyclicBarrier(num, new Runnable() {
        @Override
        public void run() {
            // TODO Auto-generated method stub
            System.out.println("go on together!");
        }
    });
    for (int i = 1; i <= num; i++) {
        new Thread(new CyclicBarrierWorker(i, barrier)).start();
    }
}

@Test
public void getId32() throws Exception {

    int nThreads = 1000;
    int size = 1000;
    CyclicBarrier cyclicBarrier = new CyclicBarrier(nThreads + 1);
    StopWatch stopWatch = new StopWatch();
    ExecutorService executorService = Executors.newFixedThreadPool(nThreads);
    stopWatch.start();
    for (int i = 0; i < nThreads; i++) {
        int port = (8800 + (i % 10));
        executorService.execute(new Runnable() {
            @Override
            public void run() {
                try {
                    cyclicBarrier.await();
                    for (int j = 0; j < size; j++) {
                        mockMvc.perform(
                                get("/api/snowflake/get-id32?partnerKey=A" + port)
                                        .header("Content-Type", "application/json;charset=UTF-8"))
                               .andExpect(status().isOk())
                               .andExpect(jsonPath("$.code").value(0))
                               .andExpect(jsonPath("$.data.id").isNumber());
                    }
                    cyclicBarrier.await();
                } catch (Exception e) {
                    e.printStackTrace();
                }

            }
        });
    }
    cyclicBarrier.await();
    cyclicBarrier.await();
    stopWatch.stop();
    System.out.println(stopWatch.prettyPrint());
    executorService.shutdown();
}

public void testTaskThrowsError() throws Exception {
  class MyError extends Error {}
  final CyclicBarrier barrier = new CyclicBarrier(2);
  // we need to make sure the error gets thrown on a different thread.
  ExecutorService service = Executors.newSingleThreadExecutor();
  try {
    final SerializingExecutor executor = new SerializingExecutor(service);
    Runnable errorTask = new Runnable() {
      @Override
      public void run() {
        throw new MyError();
      }
    };
    Runnable barrierTask = new Runnable() {
      @Override
      public void run() {
        try {
          barrier.await();
        } catch (Exception e) {
          throw new RuntimeException(e);
        }
      }
    };
    executor.execute(errorTask);
    service.execute(barrierTask);  // submit directly to the service
    // the barrier task runs after the error task so we know that the error has been observed by
    // SerializingExecutor by the time the barrier is satified
    barrier.await(10, TimeUnit.SECONDS);
    executor.execute(barrierTask);
    // timeout means the second task wasn't even tried
    barrier.await(10, TimeUnit.SECONDS);
  } finally {
    service.shutdown();
  }
}

private void waitForBarrier(CyclicBarrier b) {
    try {
        b.await();
    } catch (InterruptedException | BrokenBarrierException e) {
        Assert.fail("Test error: Caught unexpected exception:", e);
    }
}

public void testConcurrentPrimary() throws InterruptedException {
    Thread[] threads = new Thread[randomIntBetween(2, 5)];
    final int opsPerThread = randomIntBetween(10, 20);
    final int maxOps = opsPerThread * threads.length;
    final long unFinishedSeq = randomIntBetween(0, maxOps - 2); // make sure we always index the last seqNo to simplify maxSeq checks
    logger.info("--> will run [{}] threads, maxOps [{}], unfinished seq no [{}]", threads.length, maxOps, unFinishedSeq);
    final CyclicBarrier barrier = new CyclicBarrier(threads.length);
    for (int t = 0; t < threads.length; t++) {
        final int threadId = t;
        threads[t] = new Thread(new AbstractRunnable() {
            @Override
            public void onFailure(Exception e) {
                throw new ElasticsearchException("failure in background thread", e);
            }

            @Override
            protected void doRun() throws Exception {
                barrier.await();
                for (int i = 0; i < opsPerThread; i++) {
                    long seqNo = tracker.generateSeqNo();
                    logger.info("[t{}] started   [{}]", threadId, seqNo);
                    if (seqNo != unFinishedSeq) {
                        tracker.markSeqNoAsCompleted(seqNo);
                        logger.info("[t{}] completed [{}]", threadId, seqNo);
                    }
                }
            }
        }, "testConcurrentPrimary_" + threadId);
        threads[t].start();
    }
    for (Thread thread : threads) {
        thread.join();
    }
    assertThat(tracker.getMaxSeqNo(), equalTo(maxOps - 1L));
    assertThat(tracker.getCheckpoint(), equalTo(unFinishedSeq - 1L));
    tracker.markSeqNoAsCompleted(unFinishedSeq);
    assertThat(tracker.getCheckpoint(), equalTo(maxOps - 1L));
    assertThat(tracker.processedSeqNo.size(), isOneOf(0, 1));
    assertThat(tracker.firstProcessedSeqNo, equalTo(((long) maxOps / SMALL_CHUNK_SIZE) * SMALL_CHUNK_SIZE));
}

public void testConcurrentGetAndFlush() throws Exception {
    ParsedDocument doc = testParsedDocument("1", "test", null, testDocumentWithTextField(), B_1, null);
    engine.index(indexForDoc(doc));

    final AtomicReference<Engine.GetResult> latestGetResult = new AtomicReference<>();
    latestGetResult.set(engine.get(new Engine.Get(true, newUid(doc))));
    final AtomicBoolean flushFinished = new AtomicBoolean(false);
    final CyclicBarrier barrier = new CyclicBarrier(2);
    Thread getThread = new Thread(() -> {
        try {
            barrier.await();
        } catch (InterruptedException | BrokenBarrierException e) {
            throw new RuntimeException(e);
        }
        while (flushFinished.get() == false) {
            Engine.GetResult previousGetResult = latestGetResult.get();
            if (previousGetResult != null) {
                previousGetResult.release();
            }
            latestGetResult.set(engine.get(new Engine.Get(true, newUid(doc))));
            if (latestGetResult.get().exists() == false) {
                break;
            }
        }
    });
    getThread.start();
    barrier.await();
    engine.flush();
    flushFinished.set(true);
    getThread.join();
    assertTrue(latestGetResult.get().exists());
    latestGetResult.get().release();
}

@Override
public ClientRunnable getClientRunnable(String targetIP, int targetPort, int clientNums, int rpcTimeout,
                                        CyclicBarrier barrier,
                                        CountDownLatch latch, long endTime, long startTime) {
    return new SimpleProcessorBenchmarkClientRunnable(targetIP, targetPort, clientNums, rpcTimeout,
                                                     barrier, latch, startTime, endTime);
}

@SuppressWarnings("rawtypes")
@Override
public ClientRunnable getClientRunnable(String targetIP, int targetPort, int clientNums, int rpcTimeout,
                                        CyclicBarrier barrier, CountDownLatch latch, long startTime, long endTime) throws IllegalArgumentException, SecurityException, InstantiationException, IllegalAccessException, InvocationTargetException, NoSuchMethodException, ClassNotFoundException {
    String runnable = properties.getProperty("classname");
    Class[] parameterTypes = new Class[] { String.class, int.class, int.class, int.class, CyclicBarrier.class,
            CountDownLatch.class, long.class, long.class };
    Object[] parameters = new Object[] { targetIP, targetPort, clientNums, rpcTimeout, barrier, latch, startTime,
            endTime };
       return (ClientRunnable) Class.forName(runnable).getConstructor(parameterTypes).newInstance(parameters);
}

public AbstractBenchmarkClientRunnable(BenchmarkService benchmarkService, CyclicBarrier barrier, CountDownLatch latch, long startTime, long endTime) {
    this.cyclicBarrier = barrier;
    this.countDownLatch = latch;
    this.startTime = startTime;
    this.endTime = endTime;
    this.benchmarkService = benchmarkService;

    statisticTime = (int) ((endTime - startTime) / 1000000);
    statistics = new RunnableStatistics(statisticTime);
}

/**
 * A 1-party barrier triggers after single await
 */
public void testSingleParty() throws Exception {
    CyclicBarrier b = new CyclicBarrier(1);
    assertEquals(1, b.getParties());
    assertEquals(0, b.getNumberWaiting());
    b.await();
    b.await();
    assertEquals(0, b.getNumberWaiting());
}

private void waitForSignal(CyclicBarrier cyclicBarrier) {
  try {
    cyclicBarrier.await();
  } catch (InterruptedException | BrokenBarrierException e) {
    fail(e.getMessage());
  }
}

@SuppressWarnings("rawtypes")
@Override
public ClientRunnable getClientRunnable(String targetIP, int targetPort, int clientNums, int rpcTimeout,
                                        CyclicBarrier barrier, CountDownLatch latch, long startTime, long endTime) throws IllegalArgumentException, SecurityException, InstantiationException, IllegalAccessException, InvocationTargetException, NoSuchMethodException, ClassNotFoundException {
    String runnable = properties.getProperty("classname");
    Class[] parameterTypes = new Class[] { String.class, int.class, int.class, int.class, CyclicBarrier.class,
            CountDownLatch.class, long.class, long.class };
    Object[] parameters = new Object[] { targetIP, targetPort, clientNums, rpcTimeout, barrier, latch, startTime,
            endTime };
       return (ClientRunnable) Class.forName(runnable).getConstructor(parameterTypes).newInstance(parameters);
}

/**
 * Main method of the example
 * @param args
 */
public static void main(String[] args) {

    /*
     * Initializes the bi-dimensional array of data
     *      10000 rows
     *      1000 numbers in each row
     *      Looking for number 5
     */
    final int ROWS=10000;
    final int NUMBERS=1000;
    final int SEARCH=5; 
    final int PARTICIPANTS=5;
    final int LINES_PARTICIPANT=2000;
    MatrixMock mock=new MatrixMock(ROWS, NUMBERS,SEARCH);

    // Initializes the object for the results
    Results results=new Results(ROWS);

    // Creates an Grouper object
    Grouper grouper=new Grouper(results);

    // Creates the CyclicBarrier object. It has 5 participants and, when
    // they finish, the CyclicBarrier will execute the grouper object
    CyclicBarrier barrier=new CyclicBarrier(PARTICIPANTS,grouper);

    // Creates, initializes and starts 5 Searcher objects
    Searcher searchers[]=new Searcher[PARTICIPANTS];
    for (int i=0; i<PARTICIPANTS; i++){
        searchers[i]=new Searcher(i*LINES_PARTICIPANT, (i*LINES_PARTICIPANT)+LINES_PARTICIPANT, mock, results, 5,barrier);
        Thread thread=new Thread(searchers[i]);
        thread.start();
    }
    System.out.printf("Main: The main thread has finished.\n");
}

@Override
public ClientRunnable getClientRunnable(String targetIP, int targetPort, int clientNums, int rpcTimeout,
                                        CyclicBarrier barrier,
                                        CountDownLatch latch, long endTime, long startTime) {
    return new SimpleProcessorBenchmarkClientRunnable(targetIP, targetPort, clientNums, rpcTimeout,
                                                     barrier, latch, startTime, endTime);
}

/**
 * A timeout in timed await throws TimeoutException
 */
public void testAwait3_TimeoutException() throws InterruptedException {
    final CyclicBarrier c = new CyclicBarrier(2);
    Thread t = newStartedThread(new CheckedRunnable() {
        public void realRun() throws Exception {
            long startTime = System.nanoTime();
            try {
                c.await(timeoutMillis(), MILLISECONDS);
                shouldThrow();
            } catch (TimeoutException success) {}
            assertTrue(millisElapsedSince(startTime) >= timeoutMillis());
        }});

    awaitTermination(t);
}