New Phaser Synchronizer

Phaser (Etapshchik) - powerful and flexible implementation of the pattern of synchronization Barrier . Included in JDK 7 as part of the java.util.concurrent package.

Since the documentation, as they say, you cannot figure out without a hundred grams, I will describe here the principle of operation, unobvious moments and give a few examples of use.


')
Phaser quite naturally expands the functionality of its predecessor from JDK 1.5, CyclicBarrier (you can read about it here ):

• The number of barrier participants may vary.
• The stream does not have to wait until all participants gather at the barrier. It is enough to report the readiness of their work.
• On the contrary, the flow does not necessarily have to be a member of the barrier in order to expect to overcome it.

By the way, this extension is so general that it is consistent with the third barrier contract from the standard library, CountDownLatch.


The status of the attendant includes

1. phase number (phase, synchronization cycle) | int phase
2. number of participants | int parties
3. number of participants who have declared / not declared their readiness | int arrived, unarrived
4. completion status | boolean terminated

Always true:
terminated = false → phase = [real stage number, counting is from scratch]% (2 ³¹ - 1)
parties = arrived + unarrived
0 ≤ unarrived, arrived ≤ parties

All elements of the state are equipped with getters.


0. If the worker is in the terminal state (terminated = true), he does not change; calling any control method returns immediately. Phase has a negative value, parties, arrived, unarrived - the value at the time of completion.


1. Basic control methods:

register ()	register member
arrive ()	inform the attendant about his readiness without waiting for the opening of the barrier
arriveAndAwaitAdvance ()	classic arrival at the barrier. Exact counterpart of CyclicBarrier.await ()
arriveAndDeregister ()	cancel your participation

Clearly, as in the other synchronizers from the JDK, the calling thread in control methods is not tracked, therefore terms like “participant thread” and “its registration” are conditional. That is, a loaded pistol is already in our hands, it remains to send it to the foot and pull the trigger :-) However, it is not difficult to write a wrapper that corrects this dangerous situation.


2. The barrier opens immediately after any decrease unarrived to zero. ¹ That is, including when the last participant is removed, however, when creating an “empty” stage worker (new Phaser () or new Phaser (0)), the “gate is closed”.

One way or another, the barrier can be overcome only by calling one of the methods starting at “arrive”. In the context of the stream that does this, the onAdvance (phase, parties) protected method is executed - if it returns true, the stage keeper terminates his work (terminated ← true). This mechanism allows you to manage the life cycle from the inside of the class. In the default phaser implementation, it dies just when all the participants left the barrier (parties = 0).

The opening of the barrier is a transition to a new stage: phase ← phase + 1.

How about it works:

final Phaser ph = new Phaser(4); ph.arrive(); ph.register(); new Thread() {public void run(){ ph.arriveAndAwaitAdvance(); }}.start(); ph.arrive(); ph.arrive(); ph.arriveAndDeregister(); // phase number = 1 // Thread-0 released 

Above the state is written a method that led him


3. arriveAndDeregister - it is bad to think that the stream first declares readiness, and only then it is removed, which is hinted at by the name of the method. The essence would reflect “arriveToDeregister”, “arrived to put a membership card on the table” :-) Such an approach eliminates ambiguity in understanding (see illustration above): was it called onAdvance () with the parties = 4 or 5 before going to the new stage?


4. An example of changing the completion logic in the onAdvance method. A “non-destructible” land surveyor is constructed as:

 new Phaser() { protected boolean onAdvance(int phase, int parties) { return false; } }; 

Even in the method can be performed so-called. barrier action - the composition of the results of the work of participants, etc.

More specifically, what happens during the transition to the next stage:

Using

To get from Phaser CyclicBarrier, it suffices to use the method arriveAndAwaitAdvance ().

CountDownLatch emulation is a little more complicated:

 // aka countDown() phaser.arriveAndDeregister(); //  arrive()    // aka await() if (!phaser.isTerminated()) //  phaser.awaitAdvance( phaser.getPhase() ); 

The awaitAdvance (getPhase ()) construct is safe (though not atomic) because the awaitAdvance (int phaseNumber) method returns immediately if the specified step has already been passed.


In general, the presence of a stage meter greatly simplifies life and increases the number of class applications.

Parallelization of N identical actions in a row:

 import java.util.concurrent.Phaser; public class FBIEasterEgg { static int lines = 10; static String alphabet = "abcdefghijklmnopqrstuvwxyz"; static StringBuffer randomAlphabet = new StringBuffer(); public static void main(String[] args) { final Phaser phaser = new Phaser() { protected boolean onAdvance(int phase, int parties) { // John Nash mode /* print "deviations" for (int i = 0; i < alphabet.length(); i++) { System.out.printf("%d ", randomAlphabet.indexOf( alphabet.charAt(i) + "") - i); } System.out.println(); */ System.out.println(randomAlphabet); // randomAlphabet = new StringBuffer(); return phase >= lines; //loop count managing here } }; // everyone have to wait for the main thread phaser.register(); for (int i = 0; i < alphabet.length(); i++) { final char next = alphabet.charAt(i); phaser.register(); // ticket for the following thread new Thread() { public void run() { do { randomAlphabet.append(next); phaser.arriveAndAwaitAdvance(); // checkout } while ( !phaser.isTerminated() ); } }.start(); } System.out.println("Write this by hand and carry in the pocket:"); // some additional preparations may be done here // release phaser.arriveAndDeregister(); } } 

Wait until the specified number of events occur:

 class EventCounter { private Phaser count = new Phaser(1); public void eventOccured() { count.arrive(); } public void waitFor(int events) { count.register(); for (int i = 0; i < events; i++) { count.arriveAndAwaitAdvance(); } count.arriveAndDeregister(); } } 

Implementation

Doug Lee used a clever idea: to keep all the status of the stageman in one long, and to change through atomic compare-and-set operations.

Due to this, the implementation almost does not use normal locks.

Links

Phaser javadoc
JDK 7 download page
Concurrency JSR-166 Interest Site - there is a link to download a new java.util.concurrent for JDK 6.


P. S. The name “Etapshchik” is, of course, clumsy, if someone knows / thinks up better - do not hide.
¹ - provided that this Phaser is the root of the hierarchy. In this review, the scaling mechanism is not considered (it would still lie), see javadoc.