Phaser

// ── Basic Phaser usage ───────────────────────────────────────────────
Phaser phaser = new Phaser(1);   // register main thread as party (count = 1)

int WORKERS = 4;
for (int i = 0; i < WORKERS; i++) {
    phaser.register();            // register each worker dynamically (count grows to 5)
    final int id = i;
    new Thread(() -> {
        System.out.println("Worker " + id + " phase 0: working");
        phaser.arriveAndAwaitAdvance();  // arrive + wait for all 5 parties

        System.out.println("Worker " + id + " phase 1: working");
        phaser.arriveAndDeregister();    // arrive + deregister — no more phases for this worker
    }).start();
}

// Main thread participates in phase 0, then deregisters:
phaser.arriveAndAwaitAdvance();   // main thread arrives at phase 0
System.out.println("Phase 0 complete — main thread releasing");
phaser.arriveAndDeregister();     // main deregisters (count back to 4 workers only)

// ── arrive() without waiting ──────────────────────────────────────────
Phaser asyncPhaser = new Phaser(3);   // 3 parties

new Thread(() -> {
    doExpensiveWork();
    int phase = asyncPhaser.arrive();  // arrive without waiting — returns current phase
    System.out.println("Worker arrived at phase " + phase + ", continuing other work");
    doOtherWork();   // this thread does not wait for the barrier
}).start();

// Another thread can wait for a specific phase to complete:
int next = asyncPhaser.awaitAdvance(0);   // wait for phase 0 to complete
System.out.println("Phase 0 complete, new phase: " + next);

// ── Phase number and termination ─────────────────────────────────────
Phaser p = new Phaser(2);
System.out.println("Phase: " + p.getPhase());   // 0
p.arrive(); p.arrive();    // both parties arrive → phase advances
System.out.println("Phase: " + p.getPhase());   // 1

// Phaser terminates when registered parties = 0:
Phaser terminating = new Phaser(1);
int result = terminating.arriveAndDeregister();  // last party deregisters → terminates
System.out.println("Phase after termination: " + terminating.getPhase());  // negative

// ── Introspection ─────────────────────────────────────────────────────
Phaser info = new Phaser(5);
info.arrive(); info.arrive();   // 2 parties arrive
System.out.println("Registered:  " + info.getRegisteredParties());  // 5
System.out.println("Arrived:     " + info.getArrivedParties());     // 2
System.out.println("Unarrived:   " + info.getUnarrivedParties());   // 3
System.out.println("Phase:       " + info.getPhase());              // 0 (not advanced yet)
System.out.println("Terminated:  " + info.isTerminated());          // false

// ── Custom onAdvance: limited phases ─────────────────────────────────
public class LimitedPhaser extends Phaser {
    private final int maxPhases;

    public LimitedPhaser(int parties, int maxPhases) {
        super(parties);
        this.maxPhases = maxPhases;
    }

    @Override
    protected boolean onAdvance(int phase, int registeredParties) {
        System.out.println("Phase " + phase + " complete — "
            + registeredParties + " parties remaining");
        boolean terminate = (phase >= maxPhases - 1) || (registeredParties == 0);
        if (terminate) System.out.println("Phaser terminating after phase " + phase);
        return terminate;   // true = terminate after this phase
    }
}

LimitedPhaser limited = new LimitedPhaser(3, 2);  // 3 parties, 2 phases max

for (int i = 0; i < 3; i++) {
    final int id = i;
    new Thread(() -> {
        for (int phase = 0; !limited.isTerminated(); phase++) {
            System.out.println("Worker " + id + " executing phase " + phase);
            int next = limited.arriveAndAwaitAdvance();
            if (next < 0) break;   // negative = terminated
        }
    }).start();
}

// ── Hierarchical Phaser: tree structure for 1000 threads ─────────────
public static Phaser buildHierarchicalPhaser(int totalThreads, int fanout) {
    Phaser root = new Phaser();
    buildPhaser(root, totalThreads, fanout);
    return root;
}

private static void buildPhaser(Phaser parent, int parties, int fanout) {
    if (parties <= fanout) {
        // Leaf: register parties directly with parent
        parent.bulkRegister(parties);
    } else {
        // Internal node: create child phasers
        for (int i = 0; i < parties; i += fanout) {
            int batch = Math.min(fanout, parties - i);
            // Child phaser registered as a sub-party of parent:
            Phaser child = new Phaser(parent, 0);
            buildPhaser(child, batch, fanout);
        }
    }
}

// Each leaf thread uses its leaf Phaser:
// new Phaser(parentPhaser, 0) registers the child with the parent
// When all children of a phaser advance, the child Phaser triggers parent arrival

// ── Dynamic fork: register before forking, deregister when done ───────
public static void dynamicFork(Phaser phaser, int depth) {
    phaser.register();   // register this task as a party
    System.out.println("Task registered, depth=" + depth);

    if (depth < 3) {
        // Fork two sub-tasks:
        Thread t1 = new Thread(() -> dynamicFork(phaser, depth + 1));
        Thread t2 = new Thread(() -> dynamicFork(phaser, depth + 1));
        t1.start(); t2.start();
        try { t1.join(); t2.join(); }
        catch (InterruptedException e) { Thread.currentThread().interrupt(); }
    }

    // Work done: deregister this task
    phaser.arriveAndDeregister();
    System.out.println("Task deregistered, depth=" + depth);
}

Phaser dynamicPhaser = new Phaser();  // no initial parties
dynamicFork(dynamicPhaser, 0);        // starts dynamic tree
// Phaser terminates when all dynamically registered parties deregister

// ── CountDownLatch → Phaser migration ────────────────────────────────
// BEFORE: CountDownLatch
CountDownLatch latch = new CountDownLatch(3);
for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        doWork();
        latch.countDown();
    }).start();
}
latch.await();

// AFTER: Phaser (equivalent)
Phaser phaser = new Phaser(4);  // 3 workers + 1 main thread
for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        doWork();
        phaser.arriveAndDeregister();  // arrive and deregister (like countDown)
    }).start();
}
phaser.arriveAndAwaitAdvance();   // main thread arrives and waits (like await)
phaser.arriveAndDeregister();     // main deregisters — Phaser terminates

// ── CyclicBarrier → Phaser migration ─────────────────────────────────
// BEFORE: CyclicBarrier with barrier action
CyclicBarrier cb = new CyclicBarrier(3, () -> aggregateResults());
for (int phase = 0; phase < 5; phase++) {
    for (int i = 0; i < 3; i++) {
        new Thread(() -> {
            computePhase();
            try { cb.await(); } catch (Exception e) {}
        }).start();
    }
}

// AFTER: Phaser with onAdvance
Phaser migrated = new Phaser(3) {
    @Override protected boolean onAdvance(int phase, int parties) {
        aggregateResults();   // equivalent to barrier action
        return phase >= 4;    // terminate after 5 phases (0-4)
    }
};
for (int i = 0; i < 3; i++) {
    new Thread(() -> {
        while (!migrated.isTerminated()) {
            computePhase();
            migrated.arriveAndAwaitAdvance();
        }
    }).start();
}

// ── Full comparison matrix ────────────────────────────────────────────
//
//                    CountDownLatch  CyclicBarrier   Phaser
// ──────────────────────────────────────────────────────────────────────
// Reusable?           No              Yes             Yes
// Dynamic parties?    No              No              Yes
// Symmetric?          No              Yes             Configurable
// Arrive w/o wait?    No              No              Yes (arrive())
// Custom termination? No              No              Yes (onAdvance)
// Hierarchical?       No              No              Yes (parent Phaser)
// Broken state?       No              Yes             N/A (termination)
// Phase counter?      No              No              Yes
// Best use            One-time events Iterative alg.  All complex cases
//                     Starting guns   Fixed parties   Dynamic/variable

// ── awaitAdvance for phase-specific waiting ───────────────────────────
Phaser coordinator = new Phaser(1);

// Background thread advances through phases:
new Thread(() -> {
    for (int i = 0; i < 5; i++) {
        try { Thread.sleep(200); } catch (InterruptedException e) {}
        System.out.println("Advancing to phase " + (i + 1));
        coordinator.arrive();   // advance the phaser
    }
    coordinator.arriveAndDeregister();
}).start();

// Wait only for phase 3 to complete:
int completedPhase = coordinator.awaitAdvance(3);
System.out.println("Phase 3 done, now at phase: " + completedPhase);  // 4