- /*
- * @(#)AbstractQueuedSynchronizer.java 1.2 04/02/27
- *
- * Copyright 2004 Sun Microsystems, Inc. All rights reserved.
- * SUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
- */
-
- package java.util.concurrent.locks;
- import java.util.*;
- import java.util.concurrent.*;
- import java.util.concurrent.atomic.*;
- import sun.misc.Unsafe;
-
- /**
- * Provides a framework for implementing blocking locks and related
- * synchronizers (semaphores, events, etc) that rely on
- * first-in-first-out (FIFO) wait queues. This class is designed to
- * be a useful basis for most kinds of synchronizers that rely on a
- * single atomic <tt>int</tt> value to represent state. Subclasses
- * must define the protected methods that change this state, and which
- * define what that state means in terms of this object being acquired
- * or released. Given these, the other methods in this class carry
- * out all queuing and blocking mechanics. Subclasses can maintain
- * other state fields, but only the atomically updated <tt>int</tt>
- * value manipulated using methods {@link #getState}, {@link
- * #setState} and {@link #compareAndSetState} is tracked with respect
- * to synchronization.
- *
- * <p>Subclasses should be defined as non-public internal helper
- * classes that are used to implement the synchronization properties
- * of their enclosing class. Class
- * <tt>AbstractQueuedSynchronizer</tt> does not implement any
- * synchronization interface. Instead it defines methods such as
- * {@link #acquireInterruptibly} that can be invoked as
- * appropriate by concrete locks and related synchronizers to
- * implement their public methods.
- *
- * <p>This class supports either or both a default <em>exclusive</em>
- * mode and a <em>shared</em> mode. When acquired in exclusive mode,
- * attempted acquires by other threads cannot succeed. Shared mode
- * acquires by multiple threads may (but need not) succeed. This class
- * does not "understand" these differences except in the
- * mechanical sense that when a shared mode acquire succeeds, the next
- * waiting thread (if one exists) must also determine whether it can
- * acquire as well. Threads waiting in the different modes share the
- * same FIFO queue. Usually, implementation subclasses support only
- * one of these modes, but both can come into play for example in a
- * {@link ReadWriteLock}. Subclasses that support only exclusive or
- * only shared modes need not define the methods supporting the unused mode.
- *
- * <p>This class defines a nested {@link ConditionObject} class that
- * can be used as a {@link Condition} implementation by subclasses
- * supporting exclusive mode for which method {@link
- * #isHeldExclusively} reports whether synchronization is exclusively
- * held with respect to the current thread, method {@link #release}
- * invoked with the current {@link #getState} value fully releases
- * this object, and {@link #acquire}, given this saved state value,
- * eventually restores this object to its previous acquired state. No
- * <tt>AbstractQueuedSynchronizer</tt> method otherwise creates such a
- * condition, so if this constraint cannot be met, do not use it. The
- * behavior of {@link ConditionObject} depends of course on the
- * semantics of its synchronizer implementation.
- *
- * <p> This class provides inspection, instrumentation, and monitoring
- * methods for the internal queue, as well as similar methods for
- * condition objects. These can be exported as desired into classes
- * using an <tt>AbstractQueuedSynchronizer</tt> for their
- * synchronization mechanics.
- *
- * <p> Serialization of this class stores only the underlying atomic
- * integer maintaining state, so deserialized objects have empty
- * thread queues. Typical subclasses requiring serializability will
- * define a <tt>readObject</tt> method that restores this to a known
- * initial state upon deserialization.
- *
- * <h3>Usage</h3>
- *
- * <p> To use this class as the basis of a synchronizer, redefine the
- * following methods, as applicable, by inspecting and/or modifying
- * the synchronization state using {@link #getState}, {@link
- * #setState} and/or {@link #compareAndSetState}:
- *
- * <ul>
- * <li> {@link #tryAcquire}
- * <li> {@link #tryRelease}
- * <li> {@link #tryAcquireShared}
- * <li> {@link #tryReleaseShared}
- * <li> {@link #isHeldExclusively}
- *</ul>
- *
- * Each of these methods by default throws {@link
- * UnsupportedOperationException}. Implementations of these methods
- * must be internally thread-safe, and should in general be short and
- * not block. Defining these methods is the <em>only</em> supported
- * means of using this class. All other methods are declared
- * <tt>final</tt> because they cannot be independently varied.
- *
- * <p> Even though this class is based on an internal FIFO queue, it
- * does not automatically enforce FIFO acquisition policies. The core
- * of exclusive synchronization takes the form:
- *
- * <pre>
- * Acquire:
- * while (!tryAcquire(arg)) {
- * <em>enqueue thread if it is not already queued</em>
- * <em>possibly block current thread</em>
- * }
- *
- * Release:
- * if (tryRelease(arg))
- * <em>unblock the first queued thread</em>
- * </pre>
- *
- * (Shared mode is similar but may involve cascading signals.)
- *
- * <p> Because checks in acquire are invoked before enqueuing, a newly
- * acquiring thread may <em>barge</em> ahead of others that are
- * blocked and queued. However, you can, if desired, define
- * <tt>tryAcquire</tt> and/or <tt>tryAcquireShared</tt> to disable
- * barging by internally invoking one or more of the inspection
- * methods. In particular, a strict FIFO lock can define
- * <tt>tryAcquire</tt> to immediately return <tt>false</tt> if {@link
- * #getFirstQueuedThread} does not return the current thread. A
- * normally preferable non-strict fair version can immediately return
- * <tt>false</tt> only if {@link #hasQueuedThreads} returns
- * <tt>true</tt> and <tt>getFirstQueuedThread</tt> is not the current
- * thread; or equivalently, that <tt>getFirstQueuedThread</tt> is both
- * non-null and not the current thread. Further variations are
- * possible.
- *
- * <p> Throughput and scalability are generally highest for the
- * default barging (also known as <em>greedy</em>,
- * <em>renouncement</em>, and <em>convoy-avoidance</em>) strategy.
- * While this is not guaranteed to be fair or starvation-free, earlier
- * queued threads are allowed to recontend before later queued
- * threads, and each recontention has an unbiased chance to succeed
- * against incoming threads. Also, while acquires do not
- * "spin" in the usual sense, they may perform multiple
- * invocations of <tt>tryAcquire</tt> interspersed with other
- * computations before blocking. This gives most of the benefits of
- * spins when exclusive synchronization is only briefly held, without
- * most of the liabilities when it isn't. If so desired, you can
- * augment this by preceding calls to acquire methods with
- * "fast-path" checks, possibly prechecking {@link #hasContended}
- * and/or {@link #hasQueuedThreads} to only do so if the synchronizer
- * is likely not to be contended.
- *
- * <p> This class provides an efficient and scalable basis for
- * synchronization in part by specializing its range of use to
- * synchronizers that can rely on <tt>int</tt> state, acquire, and
- * release parameters, and an internal FIFO wait queue. When this does
- * not suffice, you can build synchronizers from a lower level using
- * {@link java.util.concurrent.atomic atomic} classes, your own custom
- * {@link java.util.Queue} classes, and {@link LockSupport} blocking
- * support.
- *
- * <h3>Usage Examples</h3>
- *
- * <p>Here is a non-reentrant mutual exclusion lock class that uses
- * the value zero to represent the unlocked state, and one to
- * represent the locked state. It also supports conditions and exposes
- * one of the instrumentation methods:
- *
- * <pre>
- * class Mutex implements Lock, java.io.Serializable {
- *
- * // Our internal helper class
- * private static class Sync extends AbstractQueuedSynchronizer {
- * // Report whether in locked state
- * protected boolean isHeldExclusively() {
- * return getState() == 1;
- * }
- *
- * // Acquire the lock if state is zero
- * public boolean tryAcquire(int acquires) {
- * assert acquires == 1; // Otherwise unused
- * return compareAndSetState(0, 1);
- * }
- *
- * // Release the lock by setting state to zero
- * protected boolean tryRelease(int releases) {
- * assert releases == 1; // Otherwise unused
- * if (getState() == 0) throw new IllegalMonitorStateException();
- * setState(0);
- * return true;
- * }
- *
- * // Provide a Condition
- * Condition newCondition() { return new ConditionObject(); }
- *
- * // Deserialize properly
- * private void readObject(ObjectInputStream s) throws IOException, ClassNotFoundException {
- * s.defaultReadObject();
- * setState(0); // reset to unlocked state
- * }
- * }
- *
- * // The sync object does all the hard work. We just forward to it.
- * private final Sync sync = new Sync();
- *
- * public void lock() { sync.acquire(1); }
- * public boolean tryLock() { return sync.tryAcquire(1); }
- * public void unlock() { sync.release(1); }
- * public Condition newCondition() { return sync.newCondition(); }
- * public boolean isLocked() { return sync.isHeldExclusively(); }
- * public boolean hasQueuedThreads() { return sync.hasQueuedThreads(); }
- * public void lockInterruptibly() throws InterruptedException {
- * sync.acquireInterruptibly(1);
- * }
- * public boolean tryLock(long timeout, TimeUnit unit) throws InterruptedException {
- * return sync.tryAcquireNanos(1, unit.toNanos(timeout));
- * }
- * }
- * </pre>
- *
- * <p> Here is a latch class that is like a {@link CountDownLatch}
- * except that it only requires a single <tt>signal</tt> to
- * fire. Because a latch is non-exclusive, it uses the <tt>shared</tt>
- * acquire and release methods.
- *
- * <pre>
- * class BooleanLatch {
- *
- * private static class Sync extends AbstractQueuedSynchronizer {
- * boolean isSignalled() { return getState() != 0; }
- *
- * protected int tryAcquireShared(int ignore) {
- * return isSignalled()? 1 : -1;
- * }
- *
- * protected boolean tryReleaseShared(int ignore) {
- * setState(1);
- * return true;
- * }
- * }
- *
- * private final Sync sync = new Sync();
- * public boolean isSignalled() { return sync.isSignalled(); }
- * public void signal() { sync.releaseShared(1); }
- * public void await() throws InterruptedException {
- * sync.acquireSharedInterruptibly(1);
- * }
- * }
- *
- * </pre>
- *
- * @since 1.5
- * @author Doug Lea
- */
- public abstract class AbstractQueuedSynchronizer implements java.io.Serializable {
- private static final long serialVersionUID = 7373984972572414691L;
-
- /**
- * Creates a new <tt>AbstractQueuedSynchronizer</tt> instance
- * with initial synchronization state of zero.
- */
- protected AbstractQueuedSynchronizer() { }
-
- /**
- * Wait queue node class.
- *
- * <p> The wait queue is a variant of a "CLH" (Craig, Landin, and
- * Hagersten) lock queue. CLH locks are normally used for
- * spinlocks. We instead use them for blocking synchronizers, but
- * use the same basic tactic of holding some of the control
- * information about a thread in the predecessor of its node. A
- * "status" field in each node keeps track of whether a thread
- * should block. A node is signalled when its predecessor
- * releases. Each node of the queue otherwise serves as a
- * specific-notification-style monitor holding a single waiting
- * thread. The status field does NOT control whether threads are
- * granted locks etc though. A thread may try to acquire if it is
- * first in the queue. But being first does not guarantee success;
- * it only gives the right to contend. So the currently released
- * contender thread may need to rewait.
- *
- * <p>To enqueue into a CLH lock, you atomically splice it in as new
- * tail. To dequeue, you just set the head field.
- * <pre>
- * +------+ prev +-----+ +-----+
- * head | | <---- | | <---- | | tail
- * +------+ +-----+ +-----+
- * </pre>
- *
- * <p>Insertion into a CLH queue requires only a single atomic
- * operation on "tail", so there is a simple atomic point of
- * demarcation from unqueued to queued. Similarly, dequeing
- * involves only updating the "head". However, it takes a bit
- * more work for nodes to determine who their successors are,
- * in part to deal with possible cancellation due to timeouts
- * and interrupts.
- *
- * <p>The "prev" links (not used in original CLH locks), are mainly
- * needed to handle cancellation. If a node is cancelled, its
- * successor is (normally) relinked to a non-cancelled
- * predecessor. For explanation of similar mechanics in the case
- * of spin locks, see the papers by Scott and Scherer at
- * http://www.cs.rochester.edu/u/scott/synchronization/
- *
- * <p>We also use "next" links to implement blocking mechanics.
- * The thread id for each node is kept in its own node, so a
- * predecessor signals the next node to wake up by traversing
- * next link to determine which thread it is. Determination of
- * successor must avoid races with newly queued nodes to set
- * the "next" fields of their predecessors. This is solved
- * when necessary by checking backwards from the atomically
- * updated "tail" when a node's successor appears to be null.
- * (Or, said differently, the next-links are an optimization
- * so that we don't usually need a backward scan.)
- *
- * <p>Cancellation introduces some conservatism to the basic
- * algorithms. Since we must poll for cancellation of other
- * nodes, we can miss noticing whether a cancelled node is
- * ahead or behind us. This is dealt with by always unparking
- * successors upon cancellation, allowing them to stabilize on
- * a new predecessor.
- *
- * <p>CLH queues need a dummy header node to get started. But
- * we don't create them on construction, because it would be wasted
- * effort if there is never contention. Instead, the node
- * is constructed and head and tail pointers are set upon first
- * contention.
- *
- * <p>Threads waiting on Conditions use the same nodes, but
- * use an additional link. Conditions only need to link nodes
- * in simple (non-concurrent) linked queues because they are
- * only accessed when exclusively held. Upon await, a node is
- * inserted into a condition queue. Upon signal, the node is
- * transferred to the main queue. A special value of status
- * field is used to mark which queue a node is on.
- *
- * <p>Thanks go to Dave Dice, Mark Moir, Victor Luchangco, Bill
- * Scherer and Michael Scott, along with members of JSR-166
- * expert group, for helpful ideas, discussions, and critiques
- * on the design of this class.
- */
- static final class Node {
- /** waitStatus value to indicate thread has cancelled */
- static final int CANCELLED = 1;
- /** waitStatus value to indicate thread needs unparking */
- static final int SIGNAL = -1;
- /** waitStatus value to indicate thread is waiting on condition */
- static final int CONDITION = -2;
- /** Marker to indicate a node is waiting in shared mode */
- static final Node SHARED = new Node();
- /** Marker to indicate a node is waiting in exclusive mode */
- static final Node EXCLUSIVE = null;
-
- /**
- * Status field, taking on only the values:
- * SIGNAL: The successor of this node is (or will soon be)
- * blocked (via park), so the current node must
- * unpark its successor when it releases or
- * cancels. To avoid races, acquire methods must
- * first indicate they need a signal,
- * then retry the atomic acquire, and then,
- * on failure, block.
- * CANCELLED: Node is cancelled due to timeout or interrupt
- * Nodes never leave this state. In particular,
- * a thread with cancelled node never again blocks.
- * CONDITION: Node is currently on a condition queue
- * It will not be used as a sync queue node until
- * transferred. (Use of this value here
- * has nothing to do with the other uses
- * of the field, but simplifies mechanics.)
- * 0: None of the above
- *
- * The values are arranged numerically to simplify use.
- * Non-negative values mean that a node doesn't need to
- * signal. So, most code doesn't need to check for particular
- * values, just for sign.
- *
- * The field is initialized to 0 for normal sync nodes, and
- * CONDITION for condition nodes. It is modified only using
- * CAS.
- */
- volatile int waitStatus;
-
- /**
- * Link to predecessor node that current node/thread relies on
- * for checking waitStatus. Assigned during enqueing, and nulled
- * out (for sake of GC) only upon dequeuing. Also, upon
- * cancellation of a predecessor, we short-circuit while
- * finding a non-cancelled one, which will always exist
- * because the head node is never cancelled: A node becomes
- * head only as a result of successful acquire. A
- * cancelled thread never succeeds in acquiring, and a thread only
- * cancels itself, not any other node.
- */
- volatile Node prev;
-
- /**
- * Link to the successor node that the current node/thread
- * unparks upon release. Assigned once during enqueuing, and
- * nulled out (for sake of GC) when no longer needed. Upon
- * cancellation, we cannot adjust this field, but can notice
- * status and bypass the node if cancelled. The enq operation
- * does not assign next field of a predecessor until after
- * attachment, so seeing a null next field does not
- * necessarily mean that node is at end of queue. However, if
- * a next field appears to be null, we can scan prev's from
- * the tail to double-check.
- */
- volatile Node next;
-
- /**
- * The thread that enqueued this node. Initialized on
- * construction and nulled out after use.
- */
- volatile Thread thread;
-
- /**
- * Link to next node waiting on condition, or the special
- * value SHARED. Because condition queues are accessed only
- * when holding in exclusive mode, we just need a simple
- * linked queue to hold nodes while they are waiting on
- * conditions. They are then transferred to the queue to
- * re-acquire. And because conditions can only be exclusive,
- * we save a field by using special value to indicate shared
- * mode.
- */
- Node nextWaiter;
-
- /**
- * Returns true if node is waiting in shared mode
- */
- final boolean isShared() {
- return nextWaiter == SHARED;
- }
-
- /**
- * Returns previous node, or throws NullPointerException if
- * null. Use when predecessor cannot be null.
- * @return the predecessor of this node
- */
- final Node predecessor() throws NullPointerException {
- Node p = prev;
- if (p == null)
- throw new NullPointerException();
- else
- return p;
- }
-
- Node() { // Used to establish initial head or SHARED marker
- }
-
- Node(Thread thread, Node mode) { // Used by addWaiter
- this.nextWaiter = mode;
- this.thread = thread;
- }
-
- Node(Thread thread, int waitStatus) { // Used by Condition
- this.waitStatus = waitStatus;
- this.thread = thread;
- }
- }
-
- /**
- * Head of the wait queue, lazily initialized. Except for
- * initialization, it is modified only via method setHead. Note:
- * If head exists, its waitStatus is guaranteed not to be
- * CANCELLED.
- */
- private transient volatile Node head;
-
- /**
- * Tail of the wait queue, lazily initialized. Modified only via
- * method enq to add new wait node.
- */
- private transient volatile Node tail;
-
- /**
- * The synchronization state.
- */
- private volatile int state;
-
- /**
- * Returns the current value of synchronization state.
- * This operation has memory semantics of a <tt>volatile</tt> read.
- * @return current state value
- */
- protected final int getState() {
- return state;
- }
-
- /**
- * Sets the value of synchronization state.
- * This operation has memory semantics of a <tt>volatile</tt> write.
- * @param newState the new state value
- */
- protected final void setState(int newState) {
- state = newState;
- }
-
- /**
- * Atomically sets synchronization state to the given updated
- * value if the current state value equals the expected value.
- * This operation has memory semantics of a <tt>volatile</tt> read
- * and write.
- * @param expect the expected value
- * @param update the new value
- * @return true if successful. False return indicates that
- * the actual value was not equal to the expected value.
- */
- protected final boolean compareAndSetState(int expect, int update) {
- // See below for intrinsics setup to support this
- return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
- }
-
- // Queuing utilities
-
- /**
- * Insert node into queue, initializing if necessary. See picture above.
- * @param node the node to insert
- * @return node's predecessor
- */
- private Node enq(final Node node) {
- for (;;) {
- Node t = tail;
- if (t == null) { // Must initialize
- Node h = new Node(); // Dummy header
- h.next = node;
- node.prev = h;
- if (compareAndSetHead(h)) {
- tail = node;
- return h;
- }
- }
- else {
- node.prev = t;
- if (compareAndSetTail(t, node)) {
- t.next = node;
- return t;
- }
- }
- }
- }
-
- /**
- * Create and enq node for given thread and mode
- * @param current the thread
- * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
- * @return the new node
- */
- private Node addWaiter(Node mode) {
- Node node = new Node(Thread.currentThread(), mode);
- // Try the fast path of enq; backup to full enq on failure
- Node pred = tail;
- if (pred != null) {
- node.prev = pred;
- if (compareAndSetTail(pred, node)) {
- pred.next = node;
- return node;
- }
- }
- enq(node);
- return node;
- }
-
- /**
- * Set head of queue to be node, thus dequeuing. Called only by
- * acquire methods. Also nulls out unused fields for sake of GC
- * and to suppress unnecessary signals and traversals.
- * @param node the node
- */
- private void setHead(Node node) {
- head = node;
- node.thread = null;
- node.prev = null;
- }
-
- /**
- * Wake up node's successor, if one exists.
- * @param node the node
- */
- private void unparkSuccessor(Node node) {
- /*
- * Try to clear status in anticipation of signalling. It is
- * OK if this fails or if status is changed by waiting thread.
- */
- compareAndSetWaitStatus(node, Node.SIGNAL, 0);
-
- /*
- * Thread to unpark is held in successor, which is normally
- * just the next node. But if cancelled or apparently null,
- * traverse backwards from tail to find the actual
- * non-cancelled successor.
- */
- Thread thread;
- Node s = node.next;
- if (s != null && s.waitStatus <= 0)
- thread = s.thread;
- else {
- thread = null;
- for (s = tail; s != null && s != node; s = s.prev)
- if (s.waitStatus <= 0)
- thread = s.thread;
- }
- LockSupport.unpark(thread);
- }
-
- /**
- * Set head of queue, and check if successor may be waiting
- * in shared mode, if so propagating if propagate > 0.
- * @param pred the node holding waitStatus for node
- * @param node the node
- * @param propagate the return value from a tryAcquireShared
- */
- private void setHeadAndPropagate(Node node, int propagate) {
- setHead(node);
- if (propagate > 0 && node.waitStatus != 0) {
- /*
- * Don't bother fully figuring out successor. If it
- * looks null, call unparkSuccessor anyway to be safe.
- */
- Node s = node.next;
- if (s == null || s.isShared())
- unparkSuccessor(node);
- }
- }
-
- // Utilities for various versions of acquire
-
- /**
- * Cancel an ongoing attempt to acquire.
- * @param node the node
- */
- private void cancelAcquire(Node node) {
- if (node != null) { // Ignore if node doesn't exist
- node.thread = null;
- // Can use unconditional write instead of CAS here
- node.waitStatus = Node.CANCELLED;
- unparkSuccessor(node);
- }
- }
-
- /**
- * Checks and updates status for a node that failed to acquire.
- * Returns true if thread should block. This is the main signal
- * control in all acquire loops. Requires that pred == node.prev
- * @param pred node's predecessor holding status
- * @param node the node
- * @return true if thread should block
- */
- private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
- int s = pred.waitStatus;
- if (s < 0)
- /*
- * This node has already set status asking a release
- * to signal it, so it can safely park
- */
- return true;
- if (s > 0)
- /*
- * Predecessor was cancelled. Move up to its predecessor
- * and indicate retry.
- */
- node.prev = pred.prev;
- else
- /*
- * Indicate that we need a signal, but don't park yet. Caller
- * will need to retry to make sure it cannot acquire before
- * parking.
- */
- compareAndSetWaitStatus(pred, 0, Node.SIGNAL);
- return false;
- }
-
- /**
- * Convenience method to interrupt current thread.
- */
- private static void selfInterrupt() {
- Thread.currentThread().interrupt();
- }
-
- /**
- * Convenience method to park and then check if interrupted
- * @return true if interrupted
- */
- private static boolean parkAndCheckInterrupt() {
- LockSupport.park();
- return Thread.interrupted();
- }
-
- /*
- * Various flavors of acquire, varying in exclusive/shared and
- * control modes. Each is mostly the same, but annoyingly
- * different. Only a little bit of factoring is possible due to
- * interactions of exception mechanics (including ensuring that we
- * cancel if tryAcquire throws exception) and other control, at
- * least not without hurting performance too much.
- */
-
- /**
- * Acquire in exclusive uninterruptible mode for thread already in
- * queue. Used by condition wait methods as well as acquire.
- * @param node the node
- * @param arg the acquire argument
- * @return true if interrupted while waiting
- */
- final boolean acquireQueued(final Node node, int arg) {
- try {
- boolean interrupted = false;
- for (;;) {
- final Node p = node.predecessor();
- if (p == head && tryAcquire(arg)) {
- setHead(node);
- p.next = null; // help GC
- return interrupted;
- }
- if (shouldParkAfterFailedAcquire(p, node) &&
- parkAndCheckInterrupt())
- interrupted = true;
- }
- } catch (RuntimeException ex) {
- cancelAcquire(node);
- throw ex;
- }
- }
-
- /**
- * Acquire in exclusive interruptible mode
- * @param arg the acquire argument
- */
- private void doAcquireInterruptibly(int arg)
- throws InterruptedException {
- final Node node = addWaiter(Node.EXCLUSIVE);
- try {
- for (;;) {
- final Node p = node.predecessor();
- if (p == head && tryAcquire(arg)) {
- setHead(node);
- p.next = null; // help GC
- return;
- }
- if (shouldParkAfterFailedAcquire(p, node) &&
- parkAndCheckInterrupt())
- break;
- }
- } catch (RuntimeException ex) {
- cancelAcquire(node);
- throw ex;
- }
- // Arrive here only if interrupted
- cancelAcquire(node);
- throw new InterruptedException();
- }
-
- /**
- * Acquire in exclusive timed mode
- * @param arg the acquire argument
- * @param nanosTimeout max wait time
- * @return true if acquired
- */
- private boolean doAcquireNanos(int arg, long nanosTimeout)
- throws InterruptedException {
- long lastTime = System.nanoTime();
- final Node node = addWaiter(Node.EXCLUSIVE);
- try {
- for (;;) {
- final Node p = node.predecessor();
- if (p == head && tryAcquire(arg)) {
- setHead(node);
- p.next = null; // help GC
- return true;
- }
- if (nanosTimeout <= 0) {
- cancelAcquire(node);
- return false;
- }
- if (shouldParkAfterFailedAcquire(p, node)) {
- LockSupport.parkNanos(nanosTimeout);
- if (Thread.interrupted())
- break;
- long now = System.nanoTime();
- nanosTimeout -= now - lastTime;
- lastTime = now;
- }
- }
- } catch (RuntimeException ex) {
- cancelAcquire(node);
- throw ex;
- }
- // Arrive here only if interrupted
- cancelAcquire(node);
- throw new InterruptedException();
- }
-
- /**
- * Acquire in shared uninterruptible mode
- * @param arg the acquire argument
- */
- private void doAcquireShared(int arg) {
- final Node node = addWaiter(Node.SHARED);
- try {
- boolean interrupted = false;
- for (;;) {
- final Node p = node.predecessor();
- if (p == head) {
- int r = tryAcquireShared(arg);
- if (r >= 0) {
- setHeadAndPropagate(node, r);
- p.next = null; // help GC
- if (interrupted)
- selfInterrupt();
- return;
- }
- }
- if (shouldParkAfterFailedAcquire(p, node) &&
- parkAndCheckInterrupt())
- interrupted = true;
- }
- } catch (RuntimeException ex) {
- cancelAcquire(node);
- throw ex;
- }
- }
-
- /**
- * Acquire in shared interruptible mode
- * @param arg the acquire argument
- */
- private void doAcquireSharedInterruptibly(int arg)
- throws InterruptedException {
- final Node node = addWaiter(Node.SHARED);
- try {
- for (;;) {
- final Node p = node.predecessor();
- if (p == head) {
- int r = tryAcquireShared(arg);
- if (r >= 0) {
- setHeadAndPropagate(node, r);
- p.next = null; // help GC
- return;
- }
- }
- if (shouldParkAfterFailedAcquire(p, node) &&
- parkAndCheckInterrupt())
- break;
- }
- } catch (RuntimeException ex) {
- cancelAcquire(node);
- throw ex;
- }
- // Arrive here only if interrupted
- cancelAcquire(node);
- throw new InterruptedException();
- }
-
- /**
- * Acquire in shared timed mode
- * @param arg the acquire argument
- * @param nanosTimeout max wait time
- * @return true if acquired
- */
- private boolean doAcquireSharedNanos(int arg, long nanosTimeout)
- throws InterruptedException {
-
- long lastTime = System.nanoTime();
- final Node node = addWaiter(Node.SHARED);
- try {
- for (;;) {
- final Node p = node.predecessor();
- if (p == head) {
- int r = tryAcquireShared(arg);
- if (r >= 0) {
- setHeadAndPropagate(node, r);
- p.next = null; // help GC
- return true;
- }
- }
- if (nanosTimeout <= 0) {
- cancelAcquire(node);
- return false;
- }
- if (shouldParkAfterFailedAcquire(p, node)) {
- LockSupport.parkNanos(nanosTimeout);
- if (Thread.interrupted())
- break;
- long now = System.nanoTime();
- nanosTimeout -= now - lastTime;
- lastTime = now;
- }
- }
- } catch (RuntimeException ex) {
- cancelAcquire(node);
- throw ex;
- }
- // Arrive here only if interrupted
- cancelAcquire(node);
- throw new InterruptedException();
- }
-
- // Main exported methods
-
- /**
- * Attempts to acquire in exclusive mode. This method should query
- * if the state of the object permits it to be acquired in the
- * exclusive mode, and if so to acquire it.
- *
- * <p>This method is always invoked by the thread performing
- * acquire. If this method reports failure, the acquire method
- * may queue the thread, if it is not already queued, until it is
- * signalled by a release from some other thread. This can be used
- * to implement method {@link Lock#tryLock()}.
- *
- * <p>The default
- * implementation throws {@link UnsupportedOperationException}
- *
- * @param arg the acquire argument. This value
- * is always the one passed to an acquire method,
- * or is the value saved on entry to a condition wait.
- * The value is otherwise uninterpreted and can represent anything
- * you like.
- * @return true if successful. Upon success, this object has been
- * acquired.
- * @throws IllegalMonitorStateException if acquiring would place
- * this synchronizer in an illegal state. This exception must be
- * thrown in a consistent fashion for synchronization to work
- * correctly.
- * @throws UnsupportedOperationException if exclusive mode is not supported
- */
- protected boolean tryAcquire(int arg) {
- throw new UnsupportedOperationException();
- }
-
- /**
- * Attempts to set the state to reflect a release in exclusive
- * mode. <p>This method is always invoked by the thread
- * performing release.
- *
- * <p>The default implementation throws
- * {@link UnsupportedOperationException}
- * @param arg the release argument. This value
- * is always the one passed to a release method,
- * or the current state value upon entry to a condition wait.
- * The value is otherwise uninterpreted and can represent anything
- * you like.
- * @return <tt>true</tt> if this object is now in a fully released state,
- * so that any waiting threads may attempt to acquire; and <tt>false</tt>
- * otherwise.
- * @throws IllegalMonitorStateException if releasing would place
- * this synchronizer in an illegal state. This exception must be
- * thrown in a consistent fashion for synchronization to work
- * correctly.
- * @throws UnsupportedOperationException if exclusive mode is not supported
- */
- protected boolean tryRelease(int arg) {
- throw new UnsupportedOperationException();
- }
-
- /**
- * Attempts to acquire in shared mode. This method should query if
- * the state of the object permits it to be acquired in the shared
- * mode, and if so to acquire it.
- *
- * <p>This method is always invoked by the thread performing
- * acquire. If this method reports failure, the acquire method
- * may queue the thread, if it is not already queued, until it is
- * signalled by a release from some other thread.
- *
- * <p>The default implementation throws {@link
- * UnsupportedOperationException}
- *
- * @param arg the acquire argument. This value
- * is always the one passed to an acquire method,
- * or is the value saved on entry to a condition wait.
- * The value is otherwise uninterpreted and can represent anything
- * you like.
- * @return a negative value on failure, zero on exclusive success,
- * and a positive value if non-exclusively successful, in which
- * case a subsequent waiting thread must check
- * availability. (Support for three different return values
- * enables this method to be used in contexts where acquires only
- * sometimes act exclusively.) Upon success, this object has been
- * acquired.
- * @throws IllegalMonitorStateException if acquiring would place
- * this synchronizer in an illegal state. This exception must be
- * thrown in a consistent fashion for synchronization to work
- * correctly.
- * @throws UnsupportedOperationException if shared mode is not supported
- */
- protected int tryAcquireShared(int arg) {
- throw new UnsupportedOperationException();
- }
-
- /**
- * Attempts to set the state to reflect a release in shared mode.
- * <p>This method is always invoked by the thread performing release.
- * <p> The default implementation throws
- * {@link UnsupportedOperationException}
- * @param arg the release argument. This value
- * is always the one passed to a release method,
- * or the current state value upon entry to a condition wait.
- * The value is otherwise uninterpreted and can represent anything
- * you like.
- * @return <tt>true</tt> if this object is now in a fully released state,
- * so that any waiting threads may attempt to acquire; and <tt>false</tt>
- * otherwise.
- * @throws IllegalMonitorStateException if releasing would place
- * this synchronizer in an illegal state. This exception must be
- * thrown in a consistent fashion for synchronization to work
- * correctly.
- * @throws UnsupportedOperationException if shared mode is not supported
- */
- protected boolean tryReleaseShared(int arg) {
- throw new UnsupportedOperationException();
- }
-
- /**
- * Returns true if synchronization is held exclusively with respect
- * to the current (calling) thread. This method is invoked
- * upon each call to a non-waiting {@link ConditionObject} method.
- * (Waiting methods instead invoke {@link #release}.)
- * <p>The default implementation throws {@link
- * UnsupportedOperationException}. This method is invoked
- * internally only within {@link ConditionObject} methods, so need
- * not be defined if conditions are not used.
- *
- * @return true if synchronization is held exclusively;
- * else false
- * @throws UnsupportedOperationException if conditions are not supported
- */
- protected boolean isHeldExclusively() {
- throw new UnsupportedOperationException();
- }
-
- /**
- * Acquires in exclusive mode, ignoring interrupts. Implemented
- * by invoking at least once {@link #tryAcquire},
- * returning on success. Otherwise the thread is queued, possibly
- * repeatedly blocking and unblocking, invoking {@link
- * #tryAcquire} until success. This method can be used
- * to implement method {@link Lock#lock}
- * @param arg the acquire argument.
- * This value is conveyed to {@link #tryAcquire} but is
- * otherwise uninterpreted and can represent anything
- * you like.
- */
- public final void acquire(int arg) {
- if (!tryAcquire(arg) &&
- acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
- selfInterrupt();
- }
-
- /**
- * Acquires in exclusive mode, aborting if interrupted.
- * Implemented by first checking interrupt status, then invoking
- * at least once {@link #tryAcquire}, returning on
- * success. Otherwise the thread is queued, possibly repeatedly
- * blocking and unblocking, invoking {@link #tryAcquire}
- * until success or the thread is interrupted. This method can be
- * used to implement method {@link Lock#lockInterruptibly}
- * @param arg the acquire argument.
- * This value is conveyed to {@link #tryAcquire} but is
- * otherwise uninterpreted and can represent anything
- * you like.
- * @throws InterruptedException if the current thread is interrupted
- */
- public final void acquireInterruptibly(int arg) throws InterruptedException {
- if (Thread.interrupted())
- throw new InterruptedException();
- if (!tryAcquire(arg))
- doAcquireInterruptibly(arg);
- }
-
- /**
- * Attempts to acquire in exclusive mode, aborting if interrupted,
- * and failing if the given timeout elapses. Implemented by first
- * checking interrupt status, then invoking at least once {@link
- * #tryAcquire}, returning on success. Otherwise, the thread is
- * queued, possibly repeatedly blocking and unblocking, invoking
- * {@link #tryAcquire} until success or the thread is interrupted
- * or the timeout elapses. This method can be used to implement
- * method {@link Lock#tryLock(long, TimeUnit)}.
- * @param arg the acquire argument.
- * This value is conveyed to {@link #tryAcquire} but is
- * otherwise uninterpreted and can represent anything
- * you like.
- * @param nanosTimeout the maximum number of nanoseconds to wait
- * @return true if acquired; false if timed out
- * @throws InterruptedException if the current thread is interrupted
- */
- public final boolean tryAcquireNanos(int arg, long nanosTimeout) throws InterruptedException {
- if (Thread.interrupted())
- throw new InterruptedException();
- return tryAcquire(arg) ||
- doAcquireNanos(arg, nanosTimeout);
- }
-
- /**
- * Releases in exclusive mode. Implemented by unblocking one or
- * more threads if {@link #tryRelease} returns true.
- * This method can be used to implement method {@link Lock#unlock}
- * @param arg the release argument.
- * This value is conveyed to {@link #tryRelease} but is
- * otherwise uninterpreted and can represent anything
- * you like.
- * @return the value returned from {@link #tryRelease}
- */
- public final boolean release(int arg) {
- if (tryRelease(arg)) {
- Node h = head;
- if (h != null && h.waitStatus != 0)
- unparkSuccessor(h);
- return true;
- }
- return false;
- }
-
- /**
- * Acquires in shared mode, ignoring interrupts. Implemented by
- * first invoking at least once {@link #tryAcquireShared},
- * returning on success. Otherwise the thread is queued, possibly
- * repeatedly blocking and unblocking, invoking {@link
- * #tryAcquireShared} until success.
- * @param arg the acquire argument.
- * This value is conveyed to {@link #tryAcquireShared} but is
- * otherwise uninterpreted and can represent anything
- * you like.
- */
- public final void acquireShared(int arg) {
- if (tryAcquireShared(arg) < 0)
- doAcquireShared(arg);
- }
-
- /**
- * Acquires in shared mode, aborting if interrupted. Implemented
- * by first checking interrupt status, then invoking at least once
- * {@link #tryAcquireShared}, returning on success. Otherwise the
- * thread is queued, possibly repeatedly blocking and unblocking,
- * invoking {@link #tryAcquireShared} until success or the thread
- * is interrupted.
- * @param arg the acquire argument.
- * This value is conveyed to {@link #tryAcquireShared} but is
- * otherwise uninterpreted and can represent anything
- * you like.
- * @throws InterruptedException if the current thread is interrupted
- */
- public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
- if (Thread.interrupted())
- throw new InterruptedException();
- if (tryAcquireShared(arg) < 0)
- doAcquireSharedInterruptibly(arg);
- }
-
- /**
- * Attempts to acquire in shared mode, aborting if interrupted, and
- * failing if the given timeout elapses. Implemented by first
- * checking interrupt status, then invoking at least once {@link
- * #tryAcquireShared}, returning on success. Otherwise, the
- * thread is queued, possibly repeatedly blocking and unblocking,
- * invoking {@link #tryAcquireShared} until success or the thread
- * is interrupted or the timeout elapses.
- * @param arg the acquire argument.
- * This value is conveyed to {@link #tryAcquireShared} but is
- * otherwise uninterpreted and can represent anything
- * you like.
- * @param nanosTimeout the maximum number of nanoseconds to wait
- * @return true if acquired; false if timed out
- * @throws InterruptedException if the current thread is interrupted
- */
- public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout) throws InterruptedException {
- if (Thread.interrupted())
- throw new InterruptedException();
- return tryAcquireShared(arg) >= 0 ||
- doAcquireSharedNanos(arg, nanosTimeout);
- }
-
- /**
- * Releases in shared mode. Implemented by unblocking one or more
- * threads if {@link #tryReleaseShared} returns true.
- * @param arg the release argument.
- * This value is conveyed to {@link #tryReleaseShared} but is
- * otherwise uninterpreted and can represent anything
- * you like.
- * @return the value returned from {@link #tryReleaseShared}
- */
- public final boolean releaseShared(int arg) {
- if (tryReleaseShared(arg)) {
- Node h = head;
- if (h != null && h.waitStatus != 0)
- unparkSuccessor(h);
- return true;
- }
- return false;
- }
-
- // Queue inspection methods
-
- /**
- * Queries whether any threads are waiting to acquire. Note that
- * because cancellations due to interrupts and timeouts may occur
- * at any time, a <tt>true</tt> return does not guarantee that any
- * other thread will ever acquire.
- *
- * <p> In this implementation, this operation returns in
- * constant time.
- *
- * @return true if there may be other threads waiting to acquire
- * the lock.
- */
- public final boolean hasQueuedThreads() {
- return head != tail;
- }
-
- /**
- * Queries whether any threads have ever contended to acquire this
- * synchronizer; that is if an acquire method has ever blocked.
- *
- * <p> In this implementation, this operation returns in
- * constant time.
- *
- * @return true if there has ever been contention
- */
- public final boolean hasContended() {
- return head != null;
- }
-
- /**
- * Returns the first (longest-waiting) thread in the queue, or
- * <tt>null</tt> if no threads are currently queued.
- *
- * <p> In this implementation, this operation normally returns in
- * constant time, but may iterate upon contention if other threads are
- * concurrently modifying the queue.
- *
- * @return the first (longest-waiting) thread in the queue, or
- * <tt>null</tt> if no threads are currently queued.
- */
- public final Thread getFirstQueuedThread() {
- // handle only fast path, else relay
- return (head == tail)? null : fullGetFirstQueuedThread();
- }
-
- /**
- * Version of getFirstQueuedThread called when fastpath fails
- */
- private Thread fullGetFirstQueuedThread() {
- /*
- * This loops only if the queue changes while we read sets of
- * fields.
- */
- for (;;) {
- Node h = head;
- if (h == null) // No queue
- return null;
-
- /*
- * The first node is normally h.next. Try to get its
- * thread field, ensuring consistent reads: If thread
- * field is nulled out or s.prev is no longer head, then
- * some other thread(s) concurrently performed setHead in
- * between some of our reads, so we must reread.
- */
- Node s = h.next;
- if (s != null) {
- Thread st = s.thread;
- Node sp = s.prev;
- if (st != null && sp == head)
- return st;
- }
-
- /*
- * Head's next field might not have been set yet, or may
- * have been unset after setHead. So we must check to see
- * if tail is actually first node, in almost the same way
- * as above.
- */
- Node t = tail;
- if (t == h) // Empty queue
- return null;
-
- if (t != null) {
- Thread tt = t.thread;
- Node tp = t.prev;
- if (tt != null && tp == head)
- return tt;
- }
- }
- }
-
- /**
- * Returns true if the given thread is currently queued.
- *
- * <p> This implementation traverses the queue to determine
- * presence of the given thread.
- *
- * @param thread the thread
- * @return true if the given thread in on the queue
- * @throws NullPointerException if thread null
- */
- public final boolean isQueued(Thread thread) {
- if (thread == null)
- throw new NullPointerException();
- for (Node p = tail; p != null; p = p.prev)
- if (p.thread == thread)
- return true;
- return false;
- }
-
- // Instrumentation and monitoring methods
-
- /**
- * Returns an estimate of the number of threads waiting to
- * acquire. The value is only an estimate because the number of
- * threads may change dynamically while this method traverses
- * internal data structures. This method is designed for use in
- * monitoring system state, not for synchronization
- * control.
- *
- * @return the estimated number of threads waiting for this lock
- */
- public final int getQueueLength() {
- int n = 0;
- for (Node p = tail; p != null; p = p.prev) {
- if (p.thread != null)
- ++n;
- }
- return n;
- }
-
- /**
- * Returns a collection containing threads that may be waiting to
- * acquire. Because the actual set of threads may change
- * dynamically while constructing this result, the returned
- * collection is only a best-effort estimate. The elements of the
- * returned collection are in no particular order. This method is
- * designed to facilitate construction of subclasses that provide
- * more extensive monitoring facilities.
- * @return the collection of threads
- */
- public final Collection<Thread> getQueuedThreads() {
- ArrayList<Thread> list = new ArrayList<Thread>();
- for (Node p = tail; p != null; p = p.prev) {
- Thread t = p.thread;
- if (t != null)
- list.add(t);
- }
- return list;
- }
-
- /**
- * Returns a collection containing threads that may be waiting to
- * acquire in exclusive mode. This has the same properties
- * as {@link #getQueuedThreads} except that it only returns
- * those threads waiting due to an exclusive acquire.
- * @return the collection of threads
- */
- public final Collection<Thread> getExclusiveQueuedThreads() {
- ArrayList<Thread> list = new ArrayList<Thread>();
- for (Node p = tail; p != null; p = p.prev) {
- if (!p.isShared()) {
- Thread t = p.thread;
- if (t != null)
- list.add(t);
- }
- }
- return list;
- }
-
- /**
- * Returns a collection containing threads that may be waiting to
- * acquire in shared mode. This has the same properties
- * as {@link #getQueuedThreads} except that it only returns
- * those threads waiting due to a shared acquire.
- * @return the collection of threads
- */
- public final Collection<Thread> getSharedQueuedThreads() {
- ArrayList<Thread> list = new ArrayList<Thread>();
- for (Node p = tail; p != null; p = p.prev) {
- if (p.isShared()) {
- Thread t = p.thread;
- if (t != null)
- list.add(t);
- }
- }
- return list;
- }
-
- /**
- * Returns a string identifying this synchronizer, as well as its
- * state. The state, in brackets, includes the String "State
- * =" followed by the current value of {@link #getState}, and
- * either "nonempty" or "empty" depending on
- * whether the queue is empty.
- *
- * @return a string identifying this synchronizer, as well as its state.
- */
- public String toString() {
- int s = getState();
- String q = hasQueuedThreads()? "non" : "";
- return super.toString() +
- "[State = " + s + ", " + q + "empty queue]";
- }
-
-
- // Internal support methods for Conditions
-
- /**
- * Returns true if a node, always one that was initially placed on
- * a condition queue, is now waiting to reacquire on sync queue.
- * @param node the node
- * @return true if is reacquiring
- */
- final boolean isOnSyncQueue(Node node) {
- if (node.waitStatus == Node.CONDITION || node.prev == null)
- return false;
- if (node.next != null) // If has successor, it must be on queue
- return true;
- /*
- * node.prev can be non-null, but not yet on queue because
- * the CAS to place it on queue can fail. So we have to
- * traverse from tail to make sure it actually made it. It
- * will always be near the tail in calls to this method, and
- * unless the CAS failed (which is unlikely), it will be
- * there, so we hardly ever traverse much.
- */
- return findNodeFromTail(node);
- }
-
- /**
- * Returns true if node is on sync queue by searching backwards from tail.
- * Called only when needed by isOnSyncQueue.
- * @return true if present
- */
- private boolean findNodeFromTail(Node node) {
- Node t = tail;
- for (;;) {
- if (t == node)
- return true;
- if (t == null)
- return false;
- t = t.prev;
- }
- }
-
- /**
- * Transfers a node from a condition queue onto sync queue.
- * Returns true if successful.
- * @param node the node
- * @return true if successfully transferred (else the node was
- * cancelled before signal).
- */
- final boolean transferForSignal(Node node) {
- /*
- * If cannot change waitStatus, the node has been cancelled.
- */
- if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
- return false;
-
- /*
- * Splice onto queue and try to set waitStatus of predecessor to
- * indicate that thread is (probably) waiting. If cancelled or
- * attempt to set waitStatus fails, wake up to resync (in which
- * case the waitStatus can be transiently and harmlessly wrong).
- */
- Node p = enq(node);
- int c = p.waitStatus;
- if (c > 0 || !compareAndSetWaitStatus(p, c, Node.SIGNAL))
- LockSupport.unpark(node.thread);
- return true;
- }
-
- /**
- * Transfers node, if necessary, to sync queue after a cancelled
- * wait. Returns true if thread was cancelled before being
- * signalled.
- * @param current the waiting thread
- * @param node its node
- * @return true if cancelled before the node was signalled.
- */
- final boolean transferAfterCancelledWait(Node node) {
- if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
- enq(node);
- return true;
- }
- /*
- * If we lost out to a signal(), then we can't proceed
- * until it finishes its enq(). Cancelling during an
- * incomplete transfer is both rare and transient, so just
- * spin.
- */
- while (!isOnSyncQueue(node))
- Thread.yield();
- return false;
- }
-
- /**
- * Invoke release with current state value; return saved state.
- * Cancel node and throw exception on failure.
- * @param node the condition node for this wait
- * @return previous sync state
- */
- final int fullyRelease(Node node) {
- try {
- int savedState = getState();
- if (release(savedState))
- return savedState;
- } catch(RuntimeException ex) {
- node.waitStatus = Node.CANCELLED;
- throw ex;
- }
- // reach here if release fails
- node.waitStatus = Node.CANCELLED;
- throw new IllegalMonitorStateException();
- }
-
- // Instrumentation methods for conditions
-
- /**
- * Queries whether the given ConditionObject
- * uses this synchronizer as its lock.
- * @param condition the condition
- * @return <tt>true</tt> if owned
- * @throws NullPointerException if condition null
- */
- public final boolean owns(ConditionObject condition) {
- if (condition == null)
- throw new NullPointerException();
- return condition.isOwnedBy(this);
- }
-
- /**
- * Queries whether any threads are waiting on the given condition
- * associated with this synchronizer. Note that because timeouts
- * and interrupts may occur at any time, a <tt>true</tt> return
- * does not guarantee that a future <tt>signal</tt> will awaken
- * any threads. This method is designed primarily for use in
- * monitoring of the system state.
- * @param condition the condition
- * @return <tt>true</tt> if there are any waiting threads.
- * @throws IllegalMonitorStateException if exclusive synchronization
- * is not held
- * @throws IllegalArgumentException if the given condition is
- * not associated with this synchronizer
- * @throws NullPointerException if condition null
- */
- public final boolean hasWaiters(ConditionObject condition) {
- if (!owns(condition))
- throw new IllegalArgumentException("Not owner");
- return condition.hasWaiters();
- }
-
- /**
- * Returns an estimate of the number of threads waiting on the
- * given condition associated with this synchronizer. Note that
- * because timeouts and interrupts may occur at any time, the
- * estimate serves only as an upper bound on the actual number of
- * waiters. This method is designed for use in monitoring of the
- * system state, not for synchronization control.
- * @param condition the condition
- * @return the estimated number of waiting threads.
- * @throws IllegalMonitorStateException if exclusive synchronization
- * is not held
- * @throws IllegalArgumentException if the given condition is
- * not associated with this synchronizer
- * @throws NullPointerException if condition null
- */
- public final int getWaitQueueLength(ConditionObject condition) {
- if (!owns(condition))
- throw new IllegalArgumentException("Not owner");
- return condition.getWaitQueueLength();
- }
-
- /**
- * Returns a collection containing those threads that may be
- * waiting on the given condition associated with this
- * synchronizer. Because the actual set of threads may change
- * dynamically while constructing this result, the returned
- * collection is only a best-effort estimate. The elements of the
- * returned collection are in no particular order.
- * @param condition the condition
- * @return the collection of threads
- * @throws IllegalMonitorStateException if exclusive synchronization
- * is not held
- * @throws IllegalArgumentException if the given condition is
- * not associated with this synchronizer
- * @throws NullPointerException if condition null
- */
- public final Collection<Thread> getWaitingThreads(ConditionObject condition) {
- if (!owns(condition))
- throw new IllegalArgumentException("Not owner");
- return condition.getWaitingThreads();
- }
-
- /**
- * Condition implementation for a {@link
- * AbstractQueuedSynchronizer} serving as the basis of a {@link
- * Lock} implementation.
- *
- * <p> Method documentation for this class describes mechanics,
- * not behavioral specifications from the point of view of Lock
- * and Condition users. Exported versions of this class will in
- * general need to be accompanied by documentation describing
- * condition semantics that rely on those of the associated
- * <tt>AbstractQueuedSynchronizer</tt>.
- *
- * <p> This class is Serializable, but all fields are transient,
- * so deserialized conditions have no waiters.
- */
- public class ConditionObject implements Condition, java.io.Serializable {
- private static final long serialVersionUID = 1173984872572414699L;
- /** First node of condition queue. */
- private transient Node firstWaiter;
- /** Last node of condition queue. */
- private transient Node lastWaiter;
-
- /**
- * Creates a new <tt>ConditionObject</tt> instance.
- */
- public ConditionObject() { }
-
- // Internal methods
-
- /**
- * Add a new waiter to wait queue
- * @return its new wait node
- */
- private Node addConditionWaiter() {
- Node node = new Node(Thread.currentThread(), Node.CONDITION);
- Node t = lastWaiter;
- if (t == null)
- firstWaiter = node;
- else
- t.nextWaiter = node;
- lastWaiter = node;
- return node;
- }
-
- /**
- * Remove and transfer nodes until hit non-cancelled one or
- * null. Split out from signal in part to encourage compilers
- * to inline the case of no waiters.
- * @param first (non-null) the first node on condition queue
- */
- private void doSignal(Node first) {
- do {
- if ( (firstWaiter = first.nextWaiter) == null)
- lastWaiter = null;
- first.nextWaiter = null;
- } while (!transferForSignal(first) &&
- (first = firstWaiter) != null);
- }
-
- /**
- * Remove and transfer all nodes.
- * @param first (non-null) the first node on condition queue
- */
- private void doSignalAll(Node first) {
- lastWaiter = firstWaiter = null;
- do {
- Node next = first.nextWaiter;
- first.nextWaiter = null;
- transferForSignal(first);
- first = next;
- } while (first != null);
- }
-
- // public methods
-
- /**
- * Moves the longest-waiting thread, if one exists, from the
- * wait queue for this condition to the wait queue for the
- * owning lock.
- * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
- * returns false
- */
- public final void signal() {
- if (!isHeldExclusively())
- throw new IllegalMonitorStateException();
- Node first = firstWaiter;
- if (first != null)
- doSignal(first);
- }
-
- /**
- * Moves all threads from the wait queue for this condition to
- * the wait queue for the owning lock.
- * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
- * returns false
- */
- public final void signalAll() {
- if (!isHeldExclusively())
- throw new IllegalMonitorStateException();
- Node first = firstWaiter;
- if (first != null)
- doSignalAll(first);
- }
-
- /**
- * Implements uninterruptible condition wait.
- * <ol>
- * <li> Save lock state returned by {@link #getState}
- * <li> Invoke {@link #release} with
- * saved state as argument, throwing
- * IllegalMonitorStateException if it fails.
- * <li> Block until signalled
- * <li> Reacquire by invoking specialized version of
- * {@link #acquire} with saved state as argument.
- * </ol>
- */
- public final void awaitUninterruptibly() {
- Node node = addConditionWaiter();
- int savedState = fullyRelease(node);
- boolean interrupted = false;
- while (!isOnSyncQueue(node)) {
- LockSupport.park();
- if (Thread.interrupted())
- interrupted = true;
- }
- if (acquireQueued(node, savedState) || interrupted)
- selfInterrupt();
- }
-
- /*
- * For interruptible waits, we need to track whether to throw
- * InterruptedException, if interrupted while blocked on
- * condition, versus reinterrupt current thread, if
- * interrupted while blocked waiting to re-acquire.
- */
-
- /** Mode meaning to reinterrupt on exit from wait */
- private static final int REINTERRUPT = 1;
- /** Mode meaning to throw InterruptedException on exit from wait */
- private static final int THROW_IE = -1;
-
- /**
- * Check for interrupt, returning THROW_IE if interrupted
- * before signalled, REINTERRUPT if after signalled, or
- * 0 if not interrupted.
- */
- private int checkInterruptWhileWaiting(Node node) {
- return (Thread.interrupted()) ?
- ((transferAfterCancelledWait(node))? THROW_IE : REINTERRUPT) :
- 0;
- }
-
- /**
- * Throw InterruptedException, reinterrupt current thread, or
- * do nothing, depending on mode.
- */
- private void reportInterruptAfterWait(int interruptMode)
- throws InterruptedException {
- if (interruptMode == THROW_IE)
- throw new InterruptedException();
- else if (interruptMode == REINTERRUPT)
- Thread.currentThread().interrupt();
- }
-
- /**
- * Implements interruptible condition wait.
- * <ol>
- * <li> If current thread is interrupted, throw InterruptedException
- * <li> Save lock state returned by {@link #getState}
- * <li> Invoke {@link #release} with
- * saved state as argument, throwing
- * IllegalMonitorStateException if it fails.
- * <li> Block until signalled or interrupted
- * <li> Reacquire by invoking specialized version of
- * {@link #acquire} with saved state as argument.
- * <li> If interrupted while blocked in step 4, throw exception
- * </ol>
- */
- public final void await() throws InterruptedException {
- if (Thread.interrupted())
- throw new InterruptedException();
- Node node = addConditionWaiter();
- int savedState = fullyRelease(node);
- int interruptMode = 0;
- while (!isOnSyncQueue(node)) {
- LockSupport.park();
- if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
- break;
- }
- if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
- interruptMode = REINTERRUPT;
- if (interruptMode != 0)
- reportInterruptAfterWait(interruptMode);
- }
-
- /**
- * Implements timed condition wait.
- * <ol>
- * <li> If current thread is interrupted, throw InterruptedException
- * <li> Save lock state returned by {@link #getState}
- * <li> Invoke {@link #release} with
- * saved state as argument, throwing
- * IllegalMonitorStateException if it fails.
- * <li> Block until signalled, interrupted, or timed out
- * <li> Reacquire by invoking specialized version of
- * {@link #acquire} with saved state as argument.
- * <li> If interrupted while blocked in step 4, throw InterruptedException
- * </ol>
- */
- public final long awaitNanos(long nanosTimeout) throws InterruptedException {
- if (Thread.interrupted())
- throw new InterruptedException();
- Node node = addConditionWaiter();
- int savedState = fullyRelease(node);
- long lastTime = System.nanoTime();
- int interruptMode = 0;
- while (!isOnSyncQueue(node)) {
- if (nanosTimeout <= 0L) {
- transferAfterCancelledWait(node);
- break;
- }
- LockSupport.parkNanos(nanosTimeout);
- if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
- break;
- long now = System.nanoTime();
- nanosTimeout -= now - lastTime;
- lastTime = now;
- }
- if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
- interruptMode = REINTERRUPT;
- if (interruptMode != 0)
- reportInterruptAfterWait(interruptMode);
- return nanosTimeout - (System.nanoTime() - lastTime);
- }
-
- /**
- * Implements absolute timed condition wait.
- * <ol>
- * <li> If current thread is interrupted, throw InterruptedException
- * <li> Save lock state returned by {@link #getState}
- * <li> Invoke {@link #release} with
- * saved state as argument, throwing
- * IllegalMonitorStateException if it fails.
- * <li> Block until signalled, interrupted, or timed out
- * <li> Reacquire by invoking specialized version of
- * {@link #acquire} with saved state as argument.
- * <li> If interrupted while blocked in step 4, throw InterruptedException
- * <li> If timed out while blocked in step 4, return false, else true
- * </ol>
- */
- public final boolean awaitUntil(Date deadline) throws InterruptedException {
- if (deadline == null)
- throw new NullPointerException();
- long abstime = deadline.getTime();
- if (Thread.interrupted())
- throw new InterruptedException();
- Node node = addConditionWaiter();
- int savedState = fullyRelease(node);
- boolean timedout = false;
- int interruptMode = 0;
- while (!isOnSyncQueue(node)) {
- if (System.currentTimeMillis() > abstime) {
- timedout = transferAfterCancelledWait(node);
- break;
- }
- LockSupport.parkUntil(abstime);
- if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
- break;
- }
- if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
- interruptMode = REINTERRUPT;
- if (interruptMode != 0)
- reportInterruptAfterWait(interruptMode);
- return !timedout;
- }
-
- /**
- * Implements timed condition wait.
- * <ol>
- * <li> If current thread is interrupted, throw InterruptedException
- * <li> Save lock state returned by {@link #getState}
- * <li> Invoke {@link #release} with
- * saved state as argument, throwing
- * IllegalMonitorStateException if it fails.
- * <li> Block until signalled, interrupted, or timed out
- * <li> Reacquire by invoking specialized version of
- * {@link #acquire} with saved state as argument.
- * <li> If interrupted while blocked in step 4, throw InterruptedException
- * <li> If timed out while blocked in step 4, return false, else true
- * </ol>
- */
- public final boolean await(long time, TimeUnit unit) throws InterruptedException {
- if (unit == null)
- throw new NullPointerException();
- long nanosTimeout = unit.toNanos(time);
- if (Thread.interrupted())
- throw new InterruptedException();
- Node node = addConditionWaiter();
- int savedState = fullyRelease(node);
- long lastTime = System.nanoTime();
- boolean timedout = false;
- int interruptMode = 0;
- while (!isOnSyncQueue(node)) {
- if (nanosTimeout <= 0L) {
- timedout = transferAfterCancelledWait(node);
- break;
- }
- LockSupport.parkNanos(nanosTimeout);
- if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
- break;
- long now = System.nanoTime();
- nanosTimeout -= now - lastTime;
- lastTime = now;
- }
- if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
- interruptMode = REINTERRUPT;
- if (interruptMode != 0)
- reportInterruptAfterWait(interruptMode);
- return !timedout;
- }
-
- // support for instrumentation
-
- /**
- * Returns true if this condition was created by the given
- * synchronization object
- * @return true if owned
- */
- final boolean isOwnedBy(AbstractQueuedSynchronizer sync) {
- return sync == AbstractQueuedSynchronizer.this;
- }
-
- /**
- * Queries whether any threads are waiting on this condition.
- * Implements {@link AbstractQueuedSynchronizer#hasWaiters}
- * @return <tt>true</tt> if there are any waiting threads.
- * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
- * returns false
- */
- protected final boolean hasWaiters() {
- if (!isHeldExclusively())
- throw new IllegalMonitorStateException();
- for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
- if (w.waitStatus == Node.CONDITION)
- return true;
- }
- return false;
- }
-
- /**
- * Returns an estimate of the number of threads waiting on
- * this condition.
- * Implements {@link AbstractQueuedSynchronizer#getWaitQueueLength}
- * @return the estimated number of waiting threads.
- * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
- * returns false
- */
- protected final int getWaitQueueLength() {
- if (!isHeldExclusively())
- throw new IllegalMonitorStateException();
- int n = 0;
- for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
- if (w.waitStatus == Node.CONDITION)
- ++n;
- }
- return n;
- }
-
- /**
- * Returns a collection containing those threads that may be
- * waiting on this Condition.
- * Implements {@link AbstractQueuedSynchronizer#getWaitingThreads}
- * @return the collection of threads
- * @throws IllegalMonitorStateException if {@link #isHeldExclusively}
- * returns false
- */
- protected final Collection<Thread> getWaitingThreads() {
- if (!isHeldExclusively())
- throw new IllegalMonitorStateException();
- ArrayList<Thread> list = new ArrayList<Thread>();
- for (Node w = firstWaiter; w != null; w = w.nextWaiter) {
- if (w.waitStatus == Node.CONDITION) {
- Thread t = w.thread;
- if (t != null)
- list.add(t);
- }
- }
- return list;
- }
- }
-
- /**
- * Setup to support compareAndSet. We need to natively implement
- * this here: For the sake of permitting future enhancements, we
- * cannot explicitly subclass AtomicInteger, which would be
- * efficient and useful otherwise. So, as the lesser of evils, we
- * natively implement using hotspot intrinsics API. And while we
- * are at it, we do the same for other CASable fields (which could
- * otherwise be done with atomic field updaters).
- */
- private static final Unsafe unsafe = Unsafe.getUnsafe();
- private static final long stateOffset;
- private static final long headOffset;
- private static final long tailOffset;
- private static final long waitStatusOffset;
-
- static {
- try {
- stateOffset = unsafe.objectFieldOffset
- (AbstractQueuedSynchronizer.class.getDeclaredField("state"));
- headOffset = unsafe.objectFieldOffset
- (AbstractQueuedSynchronizer.class.getDeclaredField("head"));
- tailOffset = unsafe.objectFieldOffset
- (AbstractQueuedSynchronizer.class.getDeclaredField("tail"));
- waitStatusOffset = unsafe.objectFieldOffset
- (Node.class.getDeclaredField("waitStatus"));
-
- } catch(Exception ex) { throw new Error(ex); }
- }
-
- /**
- * CAS head field. Used only by enq
- */
- private final boolean compareAndSetHead(Node update) {
- return unsafe.compareAndSwapObject(this, headOffset, null, update);
- }
-
- /**
- * CAS tail field. Used only by enq
- */
- private final boolean compareAndSetTail(Node expect, Node update) {
- return unsafe.compareAndSwapObject(this, tailOffset, expect, update);
- }
-
- /**
- * CAS waitStatus field of a node.
- */
- private final static boolean compareAndSetWaitStatus(Node node,
- int expect,
- int update) {
- return unsafe.compareAndSwapInt(node, waitStatusOffset,
- expect, update);
- }
-
- }