cyclicbarrier.java

来自「SRI international 发布的OAA框架软件」· Java 代码 · 共 442 行 · 第 1/2 页

/*
 * Written by Doug Lea with assistance from members of JCP JSR-166
 * Expert Group and released to the public domain. Use, modify, and
 * redistribute this code in any way without acknowledgement.
 */

package edu.emory.mathcs.backport.java.util.concurrent;

import edu.emory.mathcs.backport.java.util.concurrent.locks.*;
import edu.emory.mathcs.backport.java.util.concurrent.helpers.*;

/**
 * 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 a fast-fail 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, even those that have not yet resumed
 * from a previous {@link #await}, will also leave abnormally via
 * {@link BrokenBarrierException} (or <tt>InterruptedException</tt> if
 * they too were interrupted at about the same time).
 *
 * @since 1.5
 * @see CountDownLatch
 *
 * @author Doug Lea
 */
public class CyclicBarrier {
    /** 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 generation number. Incremented upon barrier trip.
     * Retracted upon reset.
     */
    private long generation;

    /**
     * Breakage indicator.
     */
    private boolean broken;

    /**
     * Number of parties still waiting. Counts down from parties to 0
     * on each cycle.
     */
    private int count;

    /**
     * Updates state on barrier trip and wake up everyone.
     */
    private void nextGeneration() {
        count = parties;
        ++generation;
        trip.signalAll();
    }

    /**
     * Sets barrier as broken and wake up everyone
     */
    private void breakBarrier() {
        broken = true;
        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 {
            int index = --count;
            long g = generation;

            if (broken)
                throw new BrokenBarrierException();

            if (Thread.interrupted()) {
                breakBarrier();
                throw new InterruptedException();
            }

            if (index == 0) {  // tripped
                nextGeneration();
                boolean ranAction = false;
                try {
                    Runnable command = barrierCommand;
                    if (command != null)
                        command.run();
                    ranAction = true;
                    return 0;
                } finally {
                    if (!ranAction)
                        breakBarrier();
                }
            }

            for (;;) {
                long deadline = Utils.nanoTime() + nanos;
                try {
                    if (!timed)
                        trip.await();
                    else {
                        trip.await(nanos, TimeUnit.NANOSECONDS);
                    }
                } catch (InterruptedException ie) {
                    breakBarrier();
                    throw ie;
                }

                if (broken ||
                    g > generation) // true if a reset occurred while waiting
                    throw new BrokenBarrierException();

                if (g < generation)
                    return index;

                if (timed) {
                    nanos = deadline - Utils.nanoTime();
                    if (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 <tt>null</tt> if there is no action.
     *
     * @throws IllegalArgumentException if <tt>parties</tt> is less than 1.