Phaser - how to use it as CountDownLatch(1)?

Question

I understand I could simply use CountDownLatch directly, however, as an excercise and to understand Phaser better, I would like to use it instead of COuntDownLatch.

Thus, I would create N number of awaiters, and one thread that needs to flip the latch. All awaiters, if arrive before the flip, would block, however after the latch is counted-down, then all subsequent await() returns instantly.

WIth Phaser I am not sure how to achieve this... Barrier is easy, as we have N + 1 threads, each arriving and awaiting. However, to make sure that no thread will await after the first phase, somehow eludes me.

The only way I could come up with, which is not nice, is as follows:

Phaser p = new Phaser(1);
int phase = p.getPhase();
....
// thread that awaits()
new Thread(new Runnable() {
    ....
    p.awaitAdvance(phase)
}

And the other thread simply advances the phaser to next phase. This is not ideal, so any pointers would be appreciated.

Answer 1

TL;DR: In this context use Phaser.arriveAndDeregister() for a non-blocking signal to the waiters of the phaser, which corresponds to operation CountDownLatch.countDown().

Phaser and CountDownLatch. The first thing to clarify is that a Phaser cannot generally encode a CountDownLatch. The tasks synchronizing with a Phaser must all wait for each other (all-to-all synchronization). In a CountDownLatch, there is a group of tasks that awaits some other task to open the latch.

Phaser and CyclicBarrier. Both of these mechanisms are used for all-to-all synchronization. The two differences between them are: 1) a Phaser the set of tasks using the phaser may grow during the life cycle of the phaser, whereas in a CyclicBarrier the number of participants is fixed; 2) with Phasers, a task may notify other members (participants) and not wait as long as it deregisters from that phaser, whereas all tasks using the cyclic barrier can only wait-and-notify.

Encoding a CountDownLatch with a Phaser. To encode a CountDownLatch(1) with a phaser you need to keep in mind the following:

1. Number of parties = number of waiters + 1: The number of registered parties, either via new Phaser(PARTIES_COUNT) or via Phaser.register.
2. CountDown.await() = Phaser.arriveAndAwaitAdvance()
3. CountDown.countDown() = Phaser.arriveAndDeregister()

Example. Suppose you want the child task to wait for the parent's task signal. Using CountDownLatch you would write:

import java.util.concurrent.*;

class CountDownExample {
    public static void main(String[] args) throws Exception {
        CountDownLatch l = new CountDownLatch(1);
        new Thread(() -> {
            try {
                l.await();
                System.out.println("Child: running");
            } catch (InterruptedException e) {
                throw new RuntimeException(e);
            }
        }).start();
        System.out.println("Parent: waiting a bit.");
        Thread.sleep(100);
        l.countDown();
    }
}

Using a Phaser you would write:

import java.util.concurrent.*;

class PhaserExample {
    public static void main(String[] args) throws Exception {
        Phaser ph = new Phaser(2); // 2 parties = 1 signaler and 1 waiter
        new Thread(() -> {
            ph.arriveAndAwaitAdvance();
            System.out.println("Child: running");
        }).start();
        System.out.println("Parent: waiting a bit.");
        Thread.sleep(100);
        ph.arriveAndDeregister();
    }
}

You might want to look at this post for another example.

Answer 2

I don't agree with @tiago-cogumbreiro's answer. In fact, Phaser can be used to simulate behavior of CountDownLatch.

You can achieve it using Phaser's arrive() method. It won't wait for other threads to arrive. Hence, CountDownLatch behavior can be achieved.

Phaser's arriveAndAwaitAdvance() makes thread in which it is called to wait, till other threads/parties arrive. Hence, this method can be used to simulate behavior of CyclicBarrier.

Below is source code which simulates behaviour of CountDownLatch(3) using Phaser. If you use only one WorkerThread, you can simulate CountDownLatch(1)

If you replace, arrive() method with arriveAndAwaitAdvance() method, below code would simulate CyclicBarrier.

public class PhaserAsCountDownLatch {

    static Phaser phaser = new Phaser(3);

    public static void main(String[] args) {
     
        new WorkerThread2().start();
        new WorkerThread3().start();
        new WorkerThread().start();
        
        //waits till phase 0 is completed and phase is advanced to next.
        phaser.awaitAdvance(0); 
        
        System.out.println("\nFROM MAIN THREAD: PHASE 0 COMPLETED");
        System.out.println("FROM MAIN THREAD: PHASE ADVANCED TO 1");
        System.out.println("MAIN THREAD ENDS HERE\n");
    }

    static class WorkerThread extends Thread{
        @Override
        public void run() {
            for (int i = 1; i < 5; i++) {
                System.out.println("tid: " + Thread.currentThread().getId() +  ", BEFORE ARRIVING val is: " + i);
            }
        
            phaser.arrive();
        
            System.out.println("ARRIVED tid: " + Thread.currentThread().getId());
        }
    }

    static class WorkerThread2 extends Thread{
        @Override
        public void run() {
        
            //won't wait for other threads to arrive. Hence, CountDownLatch behaviour can be achieved
            phaser.arrive();
        
            System.out.println("ARRIVED tid: " + Thread.currentThread().getId());
            for (int i = 200; i < 231; i++) {
                System.out.println("tid: " + Thread.currentThread().getId() +  " AFTER ARRIVING. val is: " + i);
            }
        }
    }

    static class WorkerThread3 extends Thread{
        @Override
        public void run() {
        
            //won't wait for other threads to arrive. Hence, CountDownLatch behaviour can be achieved
            phaser.arrive();
        
            System.out.println("ARRIVED tid: " + Thread.currentThread().getId());
            for (int i = 300; i < 331; i++) {
                System.out.println("tid: " + Thread.currentThread().getId() +  " AFTER ARRIVING. val is: " + i);
            }
        }
    }
}