phaser : barrier, semaphore : conditional

There are two kinds of threads, oxygen and hydrogen. Your goal is to group these threads to form water molecules. There is a barrier where each thread has to wait until a complete molecule can be formed. Hydrogen and oxygen threads will be given releaseHydrogen and releaseOxygen methods respectively, which will allow them to pass the barrier. These threads should pass the barrier in groups of three, and they must be able to immediately bond with each other to form a water molecule. You must guarantee that all the threads from one molecule bond before any other threads from the next molecule do.

In other words:

  • If an oxygen thread arrives at the barrier when no hydrogen threads are present, it has to wait for two hydrogen threads.
  • If a hydrogen thread arrives at the barrier when no other threads are present, it has to wait for an oxygen thread and another hydrogen thread.

We don’t have to worry about matching the threads up explicitly; that is, the threads do not necessarily know which other threads they are paired up with. The key is just that threads pass the barrier in complete sets; thus, if we examine the sequence of threads that bond and divide them into groups of three, each group should contain one oxygen and two hydrogen threads.

Write synchronization code for oxygen and hydrogen molecules that enforces these constraints.

Example1:

1
2
3
Input: "HOH"
Output: "HHO"
Explanation: "HOH" and "OHH" are also valid answers.

Constraints:

  • Total length of input string will be 3n, where 1 ≤ n ≤ 20.
  • Total number of H will be 2n in the input string.
  • Total number of O will be n in the input string.

1. pure semaphore

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
import java.util.concurrent.Semaphore;

class H2O {
    Semaphore output, hydro, oxy;
    
    public H2O() {
        output = new Semaphore(1);
        hydro = new Semaphore(0);
        oxy  = new Semaphore(0);
    }

    public void hydrogen(Runnable releaseHydrogen) throws InterruptedException {
		
        // releaseHydrogen.run() outputs "H". Do not change or remove this line.
        hydro.acquire();
        releaseHydrogen.run();
        oxy.release();
    }

    public void oxygen(Runnable releaseOxygen) throws InterruptedException {
        
        // releaseOxygen.run() outputs "O". Do not change or remove this line.
        output.acquire();
        hydro.release(2);
        oxy.acquire(2);
		releaseOxygen.run();
        output.release();
    }
}

NOT GOOD because only HHO pattern would be produced. What if random order is desired?

ランダムな順序は 欲しいです !

2. Phaser

A reusable synchronization barrier, similar in functionality to CyclicBarrier and CountDownLatch but supporting more flexible usage.

Synchronization. Like a CyclicBarrier, a Phaser may be repeatedly awaited. Method arriveAndAwaitAdvance() has effect analogous to CyclicBarrier.await. Each generation of a phaser has an associated phase number. The phase number starts at zero, and advances when all parties arrive at the phaser

Constructor: (most)

1
Phaser(int parties)

Methods: (most)

1
2
3
int arrive(): Arrives at this phaser, without waiting for others to arrive.

int arriveAndAwaitAdvance(): Arrives at this phaser and awaits others.

所以arriveAndAwaitAdvance()会等待所有的phaser都到达这里，然后这个phase完成，继续往下。

In fact, using a Phaser(barrier) is the only correct way:

1
2
3
4
5
6
7
8
9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
import java.util.concurrent.Phaser;
import java.util.concurrent.Semaphore;

class H2O {
    
    private Semaphore hydro, oxy;
    private Phaser phaser;
    
    public H2O() {
        hydro = new Semaphore(2);
        oxy = new Semaphore(1);
        phaser = new Phaser(3);
    }

    public void hydrogen(Runnable releaseHydrogen) throws InterruptedException {
		
        // releaseHydrogen.run() outputs "H". Do not change or remove this line.
        hydro.acquire();
        releaseHydrogen.run();
        phaser.arriveAndAwaitAdvance(); // 类似yield型的暂停
        hydro.release();
    }

    public void oxygen(Runnable releaseOxygen) throws InterruptedException {
        
        // releaseOxygen.run() outputs "O". Do not change or remove this line.
        oxy.acquire();
		releaseOxygen.run();
        phaser.arriveAndAwaitAdvance();
        oxy.release();
    }
}