/** * A synchronization aid that allows a set of threads to all wait for * each other to reach a common barrier point. CyclicBarriers are * useful in programs involving a fixed sized party of threads that * must occasionally wait for each other. The barrier is called * <em>cyclic</em> because it can be re-used after the waiting threads * are released. * * .... * * @since 1.5 * @see CountDownLatch * * @author Doug Lea */ publicclassCyclicBarrier{ /** * Each use of the barrier is represented as a generation instance. * The generation changes whenever the barrier is tripped, or * is reset. There can be many generations associated with threads * using the barrier - due to the non-deterministic way the lock * may be allocated to waiting threads - but only one of these * can be active at a time (the one to which <tt>count</tt> applies) * and all the rest are either broken or tripped. * There need not be an active generation if there has been a break * but no subsequent reset. */ // 1 privatestaticclassGeneration{ boolean broken = false; }
/** The lock for guarding barrier entry */ // 2 privatefinal ReentrantLock lock = new ReentrantLock(); /** Condition to wait on until tripped */ // 3 privatefinal Condition trip = lock.newCondition(); /** The number of parties */ // 4 privatefinalint parties; /* The command to run when tripped */ // 5 privatefinal 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. */ // 6 privateint count;
/** * Waits until all {@linkplain #getParties parties} have invoked * <tt>await</tt> on this barrier. * * <p>If the current thread is not the last to arrive then it is * disabled for thread scheduling purposes and lies dormant until * one of the following things happens: * <ul> * <li>The last thread arrives; or * <li>Some other thread {@linkplain Thread#interrupt interrupts} * the current thread; or * <li>Some other thread {@linkplain Thread#interrupt interrupts} * one of the other waiting threads; or * <li>Some other thread times out while waiting for barrier; or * <li>Some other thread invokes {@link #reset} on this barrier. * </ul> * * <p>If the current thread: * <ul> * <li>has its interrupted status set on entry to this method; or * <li>is {@linkplain Thread#interrupt interrupted} while waiting * </ul> * then {@link InterruptedException} is thrown and the current thread's * interrupted status is cleared. * * .... * * @return the arrival index of the current thread, where index * <tt>{@link #getParties()} - 1</tt> indicates the first * to arrive and zero indicates the last to arrive * @throws InterruptedException if the current thread was interrupted * while waiting * @throws BrokenBarrierException if <em>another</em> thread was * interrupted or timed out while the current thread was * waiting, or the barrier was reset, or the barrier was * broken when {@code await} was called, or the barrier * action (if present) failed due an exception. */ // 1 publicintawait()throws InterruptedException, BrokenBarrierException { try { return dowait(false, 0L); } catch (TimeoutException toe) { thrownew Error(toe); // cannot happen; } }
/** * Waits until all {@linkplain #getParties parties} have invoked * <tt>await</tt> on this barrier, or the specified waiting time elapses. * * ..... * * @param timeout the time to wait for the barrier * @param unit the time unit of the timeout parameter * @return the arrival index of the current thread, where index * <tt>{@link #getParties()} - 1</tt> indicates the first * to arrive and zero indicates the last to arrive * @throws InterruptedException if the current thread was interrupted * while waiting * @throws TimeoutException if the specified timeout elapses * @throws BrokenBarrierException if <em>another</em> thread was * interrupted or timed out while the current thread was * waiting, or the barrier was reset, or the barrier was broken * when {@code await} was called, or the barrier action (if * present) failed due an exception */ publicintawait(long timeout, TimeUnit unit) throws InterruptedException, BrokenBarrierException, TimeoutException { return dowait(true, unit.toNanos(timeout)); }
/** * Main barrier code, covering the various policies. */ // 1 privateintdowait(boolean timed, long nanos) throws InterruptedException, BrokenBarrierException, TimeoutException { final ReentrantLock lock = this.lock; lock.lock(); try { final Generation g = generation; // 2 if (g.broken) thrownew BrokenBarrierException();
if (Thread.interrupted()) { // 3 breakBarrier(); thrownew InterruptedException(); } // 4 int index = --count; // 5 if (index == 0) { // tripped boolean ranAction = false; try { final Runnable command = barrierCommand; // 6 if (command != null) // 7 command.run(); ranAction = true; // 8 nextGeneration(); return0; } finally { // 9 if (!ranAction) breakBarrier(); } }
// loop until tripped, broken, interrupted, or timed out // 10 for (;;) { try { if (!timed) trip.await(); elseif (nanos > 0L) nanos = trip.awaitNanos(nanos); } catch (InterruptedException ie) { if (g == generation && ! g.broken) { // 11 breakBarrier(); throw ie; } else { // We're about to finish waiting even if we had not // been interrupted, so this interrupt is deemed to // "belong" to subsequent execution. // 12 Thread.currentThread().interrupt(); } }
if (g.broken) thrownew BrokenBarrierException(); // 13 if (g != generation) return index;
/** * Updates state on barrier trip and wakes up everyone. * Called only while holding lock. */ // 1 privatevoidnextGeneration(){ // signal completion of last generation // 2 trip.signalAll(); // set up next generation // 3 count = parties; // 4 generation = new Generation(); }
标注代码分析
更新栅栏状态,唤醒所有在栅栏处等待的线程。这个方法只有在持有锁的情况下被调用。
唤醒所有在栅栏处等待的线程。
重置count。
生成新的generation。
CyclicBarrier#breakBarrier()
1 2 3 4 5 6 7 8 9 10 11 12 13
/** * Sets current barrier generation as broken and wakes up everyone. * Called only while holding lock. */ // 1 privatevoidbreakBarrier(){ // 2 generation.broken = true; // 3 count = parties; // 4 trip.signalAll(); }
/** * Creates a new <tt>CyclicBarrier</tt> that will trip when the * given number of parties (threads) are waiting upon it, and which * will execute the given barrier action when the barrier is tripped, * performed by the last thread entering the barrier. * * @param parties the number of threads that must invoke {@link #await} * before the barrier is tripped * @param barrierAction the command to execute when the barrier is * tripped, or {@code null} if there is no action * @throws IllegalArgumentException if {@code parties} is less than 1 */ publicCyclicBarrier(int parties, Runnable barrierAction){ if (parties <= 0) thrownew IllegalArgumentException(); this.parties = parties; this.count = parties; this.barrierCommand = barrierAction; }
/** * Creates a new <tt>CyclicBarrier</tt> that will trip when the * given number of parties (threads) are waiting upon it, and * does not perform a predefined action when the barrier is tripped. * * @param parties the number of threads that must invoke {@link #await} * before the barrier is tripped * @throws IllegalArgumentException if {@code parties} is less than 1 */ publicCyclicBarrier(int parties){ this(parties, null); }
/** * Returns the number of parties required to trip this barrier. * * @return the number of parties required to trip this barrier */ publicintgetParties(){ return parties; }
/** * Queries if this barrier is in a broken state. * * @return {@code true} if one or more parties broke out of this * barrier due to interruption or timeout since * construction or the last reset, or a barrier action * failed due to an exception; {@code false} otherwise. */ publicbooleanisBroken(){ final ReentrantLock lock = this.lock; lock.lock(); try { return generation.broken; } finally { lock.unlock(); } }
/** * Resets the barrier to its initial state. If any parties are * currently waiting at the barrier, they will return with a * {@link BrokenBarrierException}. Note that resets <em>after</em> * a breakage has occurred for other reasons can be complicated to * carry out; threads need to re-synchronize in some other way, * and choose one to perform the reset. It may be preferable to * instead create a new barrier for subsequent use. */ publicvoidreset(){ final ReentrantLock lock = this.lock; lock.lock(); try { // 1 breakBarrier(); // break the current generation nextGeneration(); // start a new generation } finally { lock.unlock(); } }
/** * Returns the number of parties currently waiting at the barrier. * This method is primarily useful for debugging and assertions. * * @return the number of parties currently blocked in {@link #await} */ // 2 publicintgetNumberWaiting(){ final ReentrantLock lock = this.lock; lock.lock(); try { return parties - count; } finally { lock.unlock(); } }