JAVA面试备战(十二)--CyclicBarrier源码分析

private static class Generation {
    boolean broken = false;
}

/** The lock for guarding barrier entry */
private final ReentrantLock lock = new ReentrantLock();
/** Condition to wait on until tripped */
private final Condition trip = lock.newCondition();
/** The number of parties */
private final int parties;
/* The command to run when tripped */
private final Runnable barrierCommand;
/** The current generation */
private Generation generation = new Generation();

/**
 * Number of parties still waiting. Counts down from parties to 0
 * on each generation.  It is reset to parties on each new
 * generation or when broken.
 */
private int count;

private final int parties;
private int count;

private final ReentrantLock lock = new ReentrantLock();
private final Condition trip = lock.newCondition();
private Generation generation = new Generation();

private final Runnable barrierCommand;

public CyclicBarrier(int parties) {
    this(parties, null);
}

public CyclicBarrier(int parties, Runnable barrierAction) {
    if (parties <= 0) throw new IllegalArgumentException();
    this.parties = parties;
    this.count = parties;
    this.barrierCommand = barrierAction;
}

private void nextGeneration() {
    // 唤醒当前这一代中所有等待在条件队列里的线程
    trip.signalAll();
    // 恢复count值，开启新的一代
    count = parties;
    generation = new Generation();
}

private void breakBarrier() {
    // 标记broken状态
    generation.broken = true;
    // 恢复count值
    count = parties;
    // 唤醒当前这一代中所有等待在条件队列里的线程（因为栅栏已经打破了）
    trip.signalAll();
}

public void reset() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        breakBarrier();   // break the current generation
        nextGeneration(); // start a new generation
    } finally {
        lock.unlock();
    }
}

public int await() throws InterruptedException, BrokenBarrierException {
    try {
        return dowait(false, 0L);
    } catch (TimeoutException toe) {
        throw new Error(toe); // cannot happen
    }
}

public int await(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException {
    return dowait(true, unit.toNanos(timeout));
}

private int dowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException {
    final ReentrantLock lock = this.lock;
    // 所有执行await方法的线程必须是已经持有了锁，所以这里必须先获取锁
    lock.lock();
    try {
        final Generation g = generation;

        // 前面说过，调用breakBarrier会将当前“代”的broken属性设为true
        // 如果一个正在await的线程发现barrier已经被break了，则将直接抛出BrokenBarrierException异常
        if (g.broken)
            throw new BrokenBarrierException();

        // 如果当前线程被中断了，则先将栅栏打破，再抛出InterruptedException
        // 这么做的原因是，所以等待在barrier的线程都是相互等待的，如果其中一个被中断了，那其他的就不用等了。
        if (Thread.interrupted()) {
            breakBarrier();
            throw new InterruptedException();
        }

        // 当前线程已经来到了栅栏前，先将等待的线程数减一
        int index = --count;
        
        // 如果等待的线程数为0了，说明所有的parties都到齐了
        // 则可以唤醒所有等待的线程，让大家一起通过栅栏，并重置栅栏
        if (index == 0) {  // tripped
            boolean ranAction = false;
            try {
                final Runnable command = barrierCommand;
                if (command != null)
                    // 如果创建CyclicBarrier时传入了barrierCommand
                    // 说明通过栅栏前有一些额外的工作要做
                    command.run(); 
                ranAction = true;
                // 唤醒所有线程，开启新一代
                nextGeneration();
                return 0;
            } finally {
                if (!ranAction)
                    breakBarrier();
            }
        }

        // 如果count数不为0，就将当前线程挂起，直到所有的线程到齐，或者超时，或者中断发生
        for (;;) {
            try {
                // 如果没有设定超时机制，则直接调用condition的await方法
                if (!timed)
                    trip.await();  // 当前线程在这里被挂起
                else if (nanos > 0L)
                    // 如果设了超时，则等待指定的时间
                    nanos = trip.awaitNanos(nanos); // 当前线程在这里被挂起，超时时间到了就会自动唤醒
            } catch (InterruptedException ie) {
                // 执行到这里说明线程被中断了
                // 如果线程被中断时还处于当前这一“代”，并且当前这一代还没有被broken,则先打破栅栏
                if (g == generation && ! g.broken) {
                    breakBarrier();
                    throw ie;
                } else {
                    // 注意来到这里有两种情况
                    // 一种是g!=generation，说明新的一代已经产生了，所以我们没有必要处理这个中断，只要再自我中断一下就好，交给后续的人处理
                    // 一种是g.broken = true, 说明中断前栅栏已经被打破了，既然中断发生时栅栏已经被打破了，也没有必要再处理这个中断了
                    Thread.currentThread().interrupt();
                }
            }

            // 注意，执行到这里是对应于线程从await状态被唤醒了
            
            // 这里先检测broken状态，能使broken状态变为true的，只有breakBarrier()方法，到这里对应的场景是
            // 1. 其他执行await方法的线程在挂起前就被中断了
            // 2. 其他执行await方法的线程在还处于等待中时被中断了
            // 2. 最后一个到达的线程在执行barrierCommand的时候发生了错误
            // 4. reset()方法被调用
            if (g.broken)
                throw new BrokenBarrierException();

            // 如果线程被唤醒时，新一代已经被开启了，说明一切正常，直接返回
            if (g != generation)
                return index;

            // 如果是因为超时时间到了被唤醒，则打破栅栏，返回TimeoutException
            if (timed && nanos <= 0L) {
                breakBarrier();
                throw new TimeoutException();
            }
        }
    } finally {
        lock.unlock();
    }
}

public int getParties() {
    return parties;
}

public int getNumberWaiting() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return parties - count;
    } finally {
        lock.unlock();
    }
}

public boolean isBroken() {
    final ReentrantLock lock = this.lock;
    lock.lock();
    try {
        return generation.broken;
    } finally {
        lock.unlock();
    }
}

class Solver {
    final int N;
    final float[][] data;
    final CyclicBarrier barrier;

    class Worker implements Runnable {
        int myRow;

        Worker(int row) {
            myRow = row;
        }

        public void run() {
            while (!done()) {
                processRow(myRow);

                try {
                    barrier.await();
                } catch (InterruptedException ex) {
                    return;
                } catch (BrokenBarrierException ex) {
                    return;
                }
            }
        }
    }

    public Solver(float[][] matrix) {
        data = matrix;
        N = matrix.length;
        Runnable barrierAction =
                new Runnable() {
                    public void run() {
                        mergeRows(...);
                    }
                };
        barrier = new CyclicBarrier(N, barrierAction);

        List<Thread> threads = new ArrayList<Thread>(N);
        for (int i = 0; i < N; i++) {
            Thread thread = new Thread(new Worker(i));
            threads.add(thread);
            thread.start();
        }

        // wait until done
        for (Thread thread : threads)
            thread.join();
    }
}