/*
 * Licensed to the Apache Software Foundation (ASF) under one or more contributor license
 * agreements. See the NOTICE file distributed with this work for additional information regarding
 * copyright ownership. The ASF licenses this file to You under the Apache License, Version 2.0 (the
 * "License"); you may not use this file except in compliance with the License. You may obtain a
 * copy of the License at
 *
 * http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software distributed under the License
 * is distributed on an "AS IS" BASIS, WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express
 * or implied. See the License for the specific language governing permissions and limitations under
 * the License.
 */

package org.apache.geode.redis;

import java.util.Arrays;
import java.util.List;
import java.util.concurrent.BrokenBarrierException;
import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.Executors;
import java.util.concurrent.Future;
import java.util.function.Consumer;
import java.util.stream.Collectors;

public class ConcurrentLoopingThreads {
  private final int iterationCount;
  private final Consumer<Integer>[] functions;
  private ExecutorService executorService = Executors.newCachedThreadPool();
  private List<Future<?>> loopingFutures;

  @SafeVarargs
  public ConcurrentLoopingThreads(int iterationCount,
      Consumer<Integer>... functions) {
    this.iterationCount = iterationCount;
    this.functions = functions;
  }

  /**
   * Start the operations asynchronously. Use {@link #await()} to wait for completion.
   */
  public ConcurrentLoopingThreads start() {
    CyclicBarrier latch = new CyclicBarrier(functions.length);

    loopingFutures = Arrays
        .stream(functions)
        .map(r -> new LoopingThread(r, iterationCount, latch))
        .map(t -> executorService.submit(t))
        .collect(Collectors.toList());

    return this;
  }

  /**
   * Wait for all operations to complete. Will propagate the first exception thrown by any of the
   * operations.
   */
  public void await() {
    loopingFutures.forEach(loopingThread -> {
      try {
        loopingThread.get();
      } catch (InterruptedException | ExecutionException e) {
        throw new RuntimeException(e);
      }
    });
  }

  /**
   * Start operations and only return once all are complete.
   */
  public void run() {
    start();
    await();
  }

  private static class LoopingRunnable implements Runnable {
    private final Consumer<Integer> runnable;
    private final int iterationCount;
    private final CyclicBarrier barrier;

    public LoopingRunnable(Consumer<Integer> runnable, int iterationCount,
        CyclicBarrier barrier) {
      this.runnable = runnable;
      this.iterationCount = iterationCount;
      this.barrier = barrier;
    }

    @Override
    public void run() {
      try {
        barrier.await();
      } catch (InterruptedException | BrokenBarrierException e) {
        throw new RuntimeException(e);
      }
      for (int i = 0; i < iterationCount; i++) {
        runnable.accept(i);
        Thread.yield();
      }
    }
  }

  private static class LoopingThread extends Thread {
    public LoopingThread(Consumer<Integer> runnable, int iterationCount,
        CyclicBarrier barrier) {
      super(new LoopingRunnable(runnable, iterationCount, barrier));
    }
  }
}

@@ -1771,24 +1771,115 @@
 s start() %7B%0A
+    return start(false);%0A  %7D%0A%0A  private ConcurrentLoopingThreads start(boolean lockstep) %7B%0A
     CyclicBa
@@ -2045,16 +2045,26 @@
 t, latch
+, lockstep
 ))%0A     
@@ -2630,16 +2630,157 @@
   start(
+false);%0A    await();%0A  %7D%0A%0A  /**%0A   * Start operations and run each iteration in lockstep%0A   */%0A  public void runInLockstep() %7B%0A    start(true
 );%0A    a
@@ -2977,16 +2977,52 @@
 barrier;
+%0A    private final boolean lockstep;
 %0A%0A    pu
@@ -3111,32 +3111,50 @@
 cBarrier barrier
+, boolean lockstep
 ) %7B%0A      this.r
@@ -3243,24 +3243,56 @@
  = barrier;%0A
+      this.lockstep = lockstep;%0A
     %7D%0A%0A    @
@@ -3320,24 +3320,262 @@
 oid run() %7B%0A
+      waitForBarrier();%0A      for (int i = 0; i %3C iterationCount; i++) %7B%0A        if (lockstep) %7B%0A          waitForBarrier();%0A        %7D%0A        runnable.accept(i);%0A        Thread.yield();%0A      %7D%0A    %7D%0A%0A    private void waitForBarrier() %7B%0A
       try %7B%0A
@@ -3716,117 +3716,8 @@
   %7D%0A
-      for (int i = 0; i %3C iterationCount; i++) %7B%0A        runnable.accept(i);%0A        Thread.yield();%0A      %7D%0A
     
@@ -3879,16 +3879,34 @@
  barrier
+, boolean lockstep
 ) %7B%0A    
@@ -3966,16 +3966,26 @@
  barrier
+, lockstep
 ));%0A    

Add ability for ConcurrentLoopingThreads to run in lockstep (#5997)