Phaser Vs CyclicBarrier in the context of Fork/Join

Question

While trying to understand the differences between Phaser and CyclicBarrier I have come across some links Difference between Phaser and CyclicBarrier and https://www.infoq.com/news/2008/07/phasers/ I read that the Phaser is compatible with Fork/Join interface while CyclicBarrier is not, here is a code to demonstrate this:

Phaser

 public static void main(String[] args) throws InterruptedException {

        CountDownLatch countDownLatch = new CountDownLatch(1);

        Phaser phaser = new Phaser(16){
            @Override
            protected boolean onAdvance(int phase, int registeredParties) {
                return phase ==1 || super.onAdvance(phase, registeredParties);
            }
        };

        System.out.println("Available Processors: "+Runtime.getRuntime().availableProcessors());

        ExecutorService executorService = ForkJoinPool.commonPool(); // Runtime.getRuntime().availableProcessors() -1

        for (int i = 0; i < 16; i++) {
            final int count = 0;
            executorService.submit(() -> {
                while (!phaser.isTerminated()) {
                    try {
                        Thread.sleep(ThreadLocalRandom.current().nextInt(300, 2000));
                        System.out.println(Thread.currentThread().getName() + count + " ... ");
                        phaser.arriveAndAwaitAdvance();
                        System.out.println(Thread.currentThread().getName() + count + " ... continues ... ");
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                }
                countDownLatch.countDown();
            });
        }
        countDownLatch.await();
    }

CyclicBarrier

public static void main(String[] args) throws InterruptedException {

        AtomicInteger phases = new AtomicInteger();
        CountDownLatch  countDownLatch = new CountDownLatch(1);
        CyclicBarrier cyclicBarrier = new CyclicBarrier(16, () -> phases.incrementAndGet());

        ExecutorService executorService = ForkJoinPool.commonPool();

        for (int i = 0; i < 16; i++) {
            executorService.submit(() -> {
                while (phases.get() < 1) {
                    try {
                        Thread.sleep(ThreadLocalRandom.current().nextInt(300, 2000));
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    try {
                        System.out.println(Thread.currentThread().getName() + " Ok, I am waiting ");

                        cyclicBarrier.await();

                        System.out.println(Thread.currentThread().getName() + " continued it's way ... ");
                    } catch (BrokenBarrierException e) {
                        e.printStackTrace();
                    } catch (InterruptedException e) {
                        e.printStackTrace();
                    }
                    countDownLatch.countDown();
                }
            });
        }
        countDownLatch.await();
    }

Explanation:

The two codes runs a fork/join thread pool, this mean that the threads are daemon threads and this is why I use CountDownLatch. The method commonPool() will create a thread pool with threads equal to Runtime.getRuntime().availableProcessors(), mine are 12, so it will create 12 threads. Both Phaser and CyclicBarrier in the two examples define 16 parties i.e they need 16 calls to await(), in the cyclic barrier, and arriveAndAwaitAdvance() in the Phaser, to move on.

In the example with the phaser, when the 12th thread blocks the fork/join will spawn more threads, it will create more threads, hence the phaser will eventually terminate. However, with CyclicBarrier when the 12th threads reach await() the program stops and never advances, it hangs. Obviously, because the barrier needs 16 calls, to make the threads advances, and only 12 are made by the created threads. The thread pool will not create more thread to advance the CyclicBarrier as it does with the Phaser.

The question:

How does the fork/join manages to create more threads with the Phaser but not with the CyclicBarrier? Why the methods arriveAndAwaitAdvance() made the thread pool create new threads, and how, but metho await() did not cause the threadpool to create more threads?

Answer 1

The Phaser is able to do this because it internally calls ForkJoinPool.managedBlock(ManagedBlocker) when blocking a thread.

This API of the ForkJoinPool is accessible to anyone, so you can easily enhance your CyclicBarrier version to use it, and remove thread starvation. For example with something in the vibe of:

ForkJoinPool.managedBlock(new ManagedBlocker() {

    boolean isReleasable = false;

    @Override
    public boolean block() throws InterruptedException {
        try {
            cyclicBarrier.await();
        } catch (BrokenBarrierException aE) {
            throw new IllegalStateException(aE);
        }
        return isReleasable = true;
    }

    @Override
    public boolean isReleasable() {
        return isReleasable;
    }
});