/*

  * DO NOT ALTER OR REMOVE COPYRIGHT NOTICES OR THIS FILE HEADER.

  *

  * This code is free software; you can redistribute it and/or modify it

  * under the terms of the GNU General Public License version 2 only, as

  * published by the Free Software Foundation.  Oracle designates this

  * particular file as subject to the "Classpath" exception as provided

  * by Oracle in the LICENSE file that accompanied this code.

  *

  * This code is distributed in the hope that it will be useful, but WITHOUT

  * ANY WARRANTY; without even the implied warranty of MERCHANTABILITY or

  * FITNESS FOR A PARTICULAR PURPOSE.  See the GNU General Public License

  * version 2 for more details (a copy is included in the LICENSE file that

  * accompanied this code).

  *

  * You should have received a copy of the GNU General Public License version

  * 2 along with this work; if not, write to the Free Software Foundation,

  * Inc., 51 Franklin St, Fifth Floor, Boston, MA 02110-1301 USA.

  *

  * Please contact Oracle, 500 Oracle Parkway, Redwood Shores, CA 94065 USA

  * or visit www.oracle.com if you need additional information or have any

  * questions.

  */

 /*

  * This file is available under and governed by the GNU General Public

  * License version 2 only, as published by the Free Software Foundation.

  * However, the following notice accompanied the original version of this

  * file:

  *

  * Written by Doug Lea with assistance from members of JCP JSR-166

  * Expert Group and released to the public domain, as explained at

  * http://creativecommons.org/publicdomain/zero/1.0/

  */

 package java.util.concurrent;

 import java.util.concurrent.locks.*;

 /**

  * 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.

  *

  * <p>A <tt>CyclicBarrier</tt> supports an optional {@link Runnable} command

  * that is run once per barrier point, after the last thread in the party

  * arrives, but before any threads are released.

  * This <em>barrier action</em> is useful

  * for updating shared-state before any of the parties continue.

  *

  * <p><b>Sample usage:</b> Here is an example of

  *  using a barrier in a parallel decomposition design:

  * <pre>

  * 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;

  *     barrier = new CyclicBarrier(N,

  *                                 new Runnable() {

  *                                   public void run() {

  *                                     mergeRows(...);

  *                                   }

  *                                 });

  *     for (int i = 0; i < N; ++i)

  *       new Thread(new Worker(i)).start();

  *

  *     waitUntilDone();

  *   }

  * }

  * </pre>

  * Here, each worker thread processes a row of the matrix then waits at the

  * barrier until all rows have been processed. When all rows are processed

  * the supplied {@link Runnable} barrier action is executed and merges the

  * rows. If the merger

  * determines that a solution has been found then <tt>done()</tt> will return

  * <tt>true</tt> and each worker will terminate.

  *

  * <p>If the barrier action does not rely on the parties being suspended when

  * it is executed, then any of the threads in the party could execute that

  * action when it is released. To facilitate this, each invocation of

  * {@link #await} returns the arrival index of that thread at the barrier.

  * You can then choose which thread should execute the barrier action, for

  * example:

  * <pre>  if (barrier.await() == 0) {

  *     // log the completion of this iteration

  *   }</pre>

  *

  * <p>The <tt>CyclicBarrier</tt> uses an all-or-none breakage model

  * for failed synchronization attempts: If a thread leaves a barrier

  * point prematurely because of interruption, failure, or timeout, all

  * other threads waiting at that barrier point will also leave

  * abnormally via {@link BrokenBarrierException} (or

  * {@link InterruptedException} if they too were interrupted at about

  * the same time).

  *

  * <p>Memory consistency effects: Actions in a thread prior to calling

  * {@code await()}

  * <a href="package-summary.html#MemoryVisibility"><i>happen-before</i></a>

  * actions that are part of the barrier action, which in turn

  * <i>happen-before</i> actions following a successful return from the

  * corresponding {@code await()} in other threads.

  *

  * @since 1.5

  * @see CountDownLatch

  *

  * @author Doug Lea

  */

 public class CyclicBarrier {

     /**

      * 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.

      */

     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;

     /**

      * Updates state on barrier trip and wakes up everyone.

      * Called only while holding lock.

      */

     private void nextGeneration() {

         // signal completion of last generation

         trip.signalAll();

         // set up next generation

         count = parties;

         generation = new Generation();

     }

     /**

      * Sets current barrier generation as broken and wakes up everyone.

      * Called only while holding lock.

      */

     private void breakBarrier() {

         generation.broken = true;

         count = parties;

         trip.signalAll();

     }

     /**

      * Main barrier code, covering the various policies.

      */

     private int dowait(boolean timed, long nanos)

         throws InterruptedException, BrokenBarrierException,

                TimeoutException {

         final ReentrantLock lock = this.lock;

         lock.lock();

         try {

             final Generation g = generation;

             if (g.broken)

                 throw new BrokenBarrierException();

             if (Thread.interrupted()) {

                 breakBarrier();

                 throw new InterruptedException();

             }

            int index = --count;

            if (index == 0) {  // tripped

                boolean ranAction = false;

                try {

                    final Runnable command = barrierCommand;

                    if (command != null)

                        command.run();

                    ranAction = true;

                    nextGeneration();

                    return 0;

                } finally {

                    if (!ranAction)

                        breakBarrier();

                }

            }

             // loop until tripped, broken, interrupted, or timed out

             for (;;) {

                 try {

                     if (!timed)

                         trip.await();

                     else if (nanos > 0L)

                         nanos = trip.awaitNanos(nanos);

                 } catch (InterruptedException ie) {

                     if (g == generation && ! g.broken) {

                         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.

                         Thread.currentThread().interrupt();

                     }

                 }

                 if (g.broken)

                     throw new BrokenBarrierException();

                 if (g != generation)

                     return index;

                 if (timed && nanos <= 0L) {

                     breakBarrier();

                     throw new TimeoutException();

                 }

             }

         } finally {

             lock.unlock();

         }

     }

     /**

      * 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

      */

     public CyclicBarrier(int parties, Runnable barrierAction) {

         if (parties <= 0) throw new 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

      */

     public CyclicBarrier(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

      */

     public int getParties() {

         return parties;

     }

     /**

      * 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.

      *

      * <p>If the barrier is {@link #reset} while any thread is waiting,

      * or if the barrier {@linkplain #isBroken is broken} when

      * <tt>await</tt> is invoked, or while any thread is waiting, then

      * {@link BrokenBarrierException} is thrown.

      *

      * <p>If any thread is {@linkplain Thread#interrupt interrupted} while waiting,

      * then all other waiting threads will throw

      * {@link BrokenBarrierException} and the barrier is placed in the broken

      * state.

      *

      * <p>If the current thread is the last thread to arrive, and a

      * non-null barrier action was supplied in the constructor, then the

      * current thread runs the action before allowing the other threads to

      * continue.

      * If an exception occurs during the barrier action then that exception

      * will be propagated in the current thread and the barrier is placed in

      * the broken state.

      *

      * @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.

      */

     public int await() throws InterruptedException, BrokenBarrierException {

         try {

             return dowait(false, 0L);

         } catch (TimeoutException toe) {

             throw new Error(toe); // cannot happen;

         }

     }

     /**

      * Waits until all {@linkplain #getParties parties} have invoked

      * <tt>await</tt> on this barrier, or the specified waiting time elapses.

      *

      * <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>The specified timeout elapses; 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.

      *

      * <p>If the specified waiting time elapses then {@link TimeoutException}

      * is thrown. If the time is less than or equal to zero, the

      * method will not wait at all.

      *

      * <p>If the barrier is {@link #reset} while any thread is waiting,

      * or if the barrier {@linkplain #isBroken is broken} when

      * <tt>await</tt> is invoked, or while any thread is waiting, then

      * {@link BrokenBarrierException} is thrown.

      *

      * <p>If any thread is {@linkplain Thread#interrupt interrupted} while

      * waiting, then all other waiting threads will throw {@link

      * BrokenBarrierException} and the barrier is placed in the broken

      * state.

      *

      * <p>If the current thread is the last thread to arrive, and a

      * non-null barrier action was supplied in the constructor, then the

      * current thread runs the action before allowing the other threads to

      * continue.

      * If an exception occurs during the barrier action then that exception

      * will be propagated in the current thread and the barrier is placed in

      * the broken state.

      *

      * @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

      */

     public int await(long timeout, TimeUnit unit)

         throws InterruptedException,

                BrokenBarrierException,

                TimeoutException {

         return dowait(true, unit.toNanos(timeout));

     }

     /**

      * 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.

      */

     public boolean isBroken() {

         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.

      */

     public void reset() {

         final ReentrantLock lock = this.lock;

         lock.lock();

         try {

             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}

      */

     public int getNumberWaiting() {

         final ReentrantLock lock = this.lock;

         lock.lock();

         try {

             return parties - count;

         } finally {

             lock.unlock();

         }

     }

 }