Java looping Threads using CyclicBarrier

Question 1

Following @Totoro's answer, below is a little bit of example code which also incorporates the requirement "I need the threads' run() method to keep looping until a certain condition is met, pausing after every iteration to let all threads synchronise". That makes it complex pretty quick, but hopefully the program output will clarify the example code (or I should just make better examples).

import java.util.concurrent.CyclicBarrier;
import java.util.concurrent.atomic.AtomicBoolean;
import java.util.concurrent.atomic.AtomicInteger;

public class BarrierCalc implements Runnable {

public static final int CALC_THREADS = 3;

private static final AtomicBoolean runCondition = new AtomicBoolean();
private static final AtomicBoolean stopRunning = new AtomicBoolean();

public static void main(String[] args) {

    CyclicBarrier barrier = new CyclicBarrier(CALC_THREADS + 1);
    for (int i = 0; i < CALC_THREADS; i++) {
         new Thread(new BarrierCalc(barrier)).start();
    }
    try {
        runCondition.set(true);
        barrier.await();
        showln(0, "STATS!");

        barrier.await();
        showln(0, "start looping 1");
        Thread.sleep(200);
        runCondition.set(false);
        showln(0, "stop looping 1");
        barrier.await();
        runCondition.set(true);

        barrier.await();
        showln(0, "start looping 2");
        Thread.sleep(100);
        runCondition.set(false);
        showln(0, "stop looping 2");
        barrier.await();

        stopRunning.set(true);
        showln(0, "finishing");
        barrier.await();
    } catch (Exception e) {
        e.printStackTrace();
    }
}

private static final AtomicInteger calcId = new AtomicInteger();

private CyclicBarrier barrier;
private int id;

public BarrierCalc(CyclicBarrier barrier) {
    this.barrier = barrier;
    id = calcId.incrementAndGet();
}

public void run() {

    showln(id, "waiting for start");
    try {
        barrier.await(); // display stats
        barrier.await(); // start running
        int loopNumber = 0;
        while (!stopRunning.get()) {
            showln(id, "looping " + (++loopNumber));
            while (runCondition.get()) {
                Thread.sleep(10); // simulate looping
            }
            showln(id, "synchronizing " + loopNumber);
            barrier.await();
            showln(id, "synchronized " + loopNumber);
            // give main thread a chance to set stopCondition and runCondition
            barrier.await();
        }
        showln(id, "finished");
    } catch (Exception e) {
        e.printStackTrace();
    }
}

private static final long START_TIME = System.currentTimeMillis();

public static void showln(int id, String msg) {
    System.out.println((System.currentTimeMillis() - START_TIME) + "\t ID " + id + ": " + msg);
}

}

Keep in mind that program output might not be in the order expected: threads that are writing at the same time to one synchronized output (System.out) are given write-access in random order.

Question 2

The barrier.wait() will suspend the threads. The barrier is already in the main thread, it does not need another. In your algorithm above you show the threads being restarted after displaying stats. You should not need to do this. If the recently awakened threads are in a loop they will go back into the barrier.wait() again.

Question 3

You can take a look at my example where I played with CyclicBarrier.Here each worker makes some calculation and at the barrier the condition is checked. If it meets the condition than all workers stop calculations, otherwise they continue:

class Solver {
    private static final int REQUIRED_AMOUNT = 100;
    private static final int NUMBER_OF_THREADS = 4;

    AtomicInteger atomicInteger = new AtomicInteger();
    AtomicBoolean continueCalculation = new AtomicBoolean(true);
    final CyclicBarrier barrier;

    public static void main(String[] args) {
        new Solver();
    }

    class Worker implements Runnable {
        int workerId;
        Worker(int workerId) {
            this.workerId = workerId;
        }

        public void run() {
            try {
                while(continueCalculation.get()) {
                    calculate(workerId);
                    barrier.await();
                }

            } catch (Exception ex) {
                System.out.println("Finishing " + workerId);
            }
        }
    }

    public Solver() {
        Runnable barrierAction = () -> {
            if (done()) {
                continueCalculation.set(false);
            }
        };

        barrier = new CyclicBarrier(NUMBER_OF_THREADS, barrierAction);

        List<Thread> threads = new ArrayList(NUMBER_OF_THREADS);
        for (int i = 0; i < NUMBER_OF_THREADS; i++) {
            Thread thread = new Thread(new Worker(i));
            threads.add(thread);
            thread.start();
        }
    }

    private void calculate(int workerId) throws InterruptedException {
        // Some long-running calculation
        Thread.sleep(2000L);
        int r = new Random().nextInt(12);

        System.out.println("Worker #" + workerId + " added " + r +" = " + atomicInteger.addAndGet(r));
    }

    private boolean done() {
        int currentResult = atomicInteger.get();
        boolean collected = currentResult >= REQUIRED_AMOUNT;

        System.out.println("=======================================================");
        System.out.println("Checking state at the barrier: " + currentResult);
        if (collected) {
            System.out.println("Required result is reached");
        }
        System.out.println("=======================================================");

        return collected;
    }
}