diff --git a/week_03/42/AbstractQueuedSynchronizer.md b/week_03/42/AbstractQueuedSynchronizer.md new file mode 100644 index 0000000000000000000000000000000000000000..e3b8e356f5e21d41402eed50d1859345bf7d5f7c --- /dev/null +++ b/week_03/42/AbstractQueuedSynchronizer.md @@ -0,0 +1,1324 @@ + + +package java.util.concurrent.locks; +import java.util.concurrent.TimeUnit; +import java.util.ArrayList; +import java.util.Collection; +import java.util.Date; +import sun.misc.Unsafe; + + +public abstract class AbstractQueuedSynchronizer + extends AbstractOwnableSynchronizer + implements java.io.Serializable { + + private static final long serialVersionUID = 7373984972572414691L; + + + protected AbstractQueuedSynchronizer() { } + + + static final class Node { + // 标识一个共享模式的节点 + static final Node SHARED = new Node(); + // 标识一个互斥模式的节点 + static final Node EXCLUSIVE = null; + // 标识着线程取消状态 + static final int CANCELLED = 1; + // 标识着后继线程需要被唤醒 + static final int SIGNAL = -1; + // 标识着线程等待一个条件上 + static final int CONDITION = -2; + // 标识着后面的共享锁需要无条件传播 + static final int PROPAGATE = -3; + /** + * The field is initialized to 0 for normal sync nodes, and + * CONDITION for condition nodes. It is modified using CAS + * (or when possible, unconditional volatile writes). + * 当前节点保存的线程对应的等待状态 + */ + volatile int waitStatus; + // 前节点 + volatile Node prev; + // 后节点 + volatile Node next; + // 当前节点的线程 + volatile Thread thread; + // 下一个等待在条件上的节点(Condition锁使用) + Node nextWaiter; + // 是否是共享模式 + final boolean isShared() { + return nextWaiter == SHARED; + } + // 获取前一个节点,为空则抛出异常 + 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 + // 把共享模式还是互斥模式存储到nextWaiter这个字段里面了 + this.nextWaiter = mode; + this.thread = thread; + } + + Node(Thread thread, int waitStatus) { // Used by Condition + this.waitStatus = waitStatus; + this.thread = thread; + } + } + + + // 使用unsafe方法对状态进行修改 + 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; + private static final long nextOffset; + + 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")); + nextOffset = unsafe.objectFieldOffset + (Node.class.getDeclaredField("next")); + + } catch (Exception ex) { throw new Error(ex); } + } + + // 队列头节点 + private transient volatile Node head; + + // 队列尾节点 + private transient volatile Node tail; + + // 控制加锁解锁中台的变量 + private volatile int state; + // 获取当前同步状态 + protected final int getState() { + return state; + } + // 设置同步状态 + protected final void setState(int newState) { + state = newState; + } + // 原子的操作设置状态 + 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 + + /** + * The number of nanoseconds for which it is faster to spin + * rather than to use timed park. A rough estimate suffices + * to improve responsiveness with very short timeouts. + */ + static final long spinForTimeoutThreshold = 1000L; + + // 插入队列 + private Node enq(final Node node) { + for (;;) { // 自旋不停尝试 + Node t = tail; + // 尾节点如果为空 + if (t == null) { // Must initialize + // 设置头结点 + if (compareAndSetHead(new Node())) + // 头节点也是尾节点 + tail = head; + } else { + // 待插入的节点前一个节点是尾节点 + node.prev = t; + // 重新设置尾节点 + if (compareAndSetTail(t, node)) { + // 原始尾节点的下一个节点是当前节点 + t.next = node; + // 返回原始尾节点 + return t; + } + } + } + } + + + /** + * 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); + } + + /** + * Creates and enqueues node for current thread and given mode. + * 给当前线程创建一个节点并且赋予模式(共享还是互斥模式) + * @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 + // 快速如队列法,如果失败的话则调用enq方法进行入队列 + Node pred = tail; // 队尾指针 + if (pred != null) { // 如果队尾不为空 + node.prev = pred; // 当前节点的前一个节点指向原始尾节点 + // cas 操作设置当前节点为尾节点 + if (compareAndSetTail(pred, node)) { + pred.next = node; // 设置成功的话,则原始尾节点的下一个节点指向当前节点 + return node; // 返回当前节点 + } + } + // 上述操作失败,则通过自旋方式进行入队列 + enq(node); + return node; + } + + /** + * Sets 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; + } + + + /** + * 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. + * 在互斥模式下获取锁 + */ + protected boolean tryAcquire(int arg) { + throw new UnsupportedOperationException(); + } + + /** + * Attempts to set the state to reflect a release in exclusive + * mode. + * 在互斥模式下释放锁 + */ + 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. + * 共享模式下获取锁 + */ + protected int tryAcquireShared(int arg) { + throw new UnsupportedOperationException(); + } + + /** + * Attempts to set the state to reflect a release in shared mode. + * 共享模式想释放锁 + */ + protected boolean tryReleaseShared(int arg) { + throw new UnsupportedOperationException(); + } + + /** + * Returns {@code 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}.) + * 如果当前线程占有锁则返回true + */ + protected boolean isHeldExclusively() { + throw new UnsupportedOperationException(); + } + + /** + * Wakes up node's successor, if one exists. + * + * @param node the node + */ + private void unparkSuccessor(Node node) { + /* + * If status is negative (i.e., possibly needing signal) try + * to clear in anticipation of signalling. It is OK if this + * fails or if status is changed by waiting thread. + */ + int ws = node.waitStatus; + if (ws < 0) + compareAndSetWaitStatus(node, ws, 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. + */ + Node s = node.next; + if (s == null || s.waitStatus > 0) { + s = null; + for (Node t = tail; t != null && t != node; t = t.prev) + if (t.waitStatus <= 0) + s = t; + } + if (s != null) + LockSupport.unpark(s.thread); + } + + /** + * Release action for shared mode -- signals successor and ensures + * propagation. (Note: For exclusive mode, release just amounts + * to calling unparkSuccessor of head if it needs signal.) + */ + private void doReleaseShared() { + /* + * Ensure that a release propagates, even if there are other + * in-progress acquires/releases. This proceeds in the usual + * way of trying to unparkSuccessor of head if it needs + * signal. But if it does not, status is set to PROPAGATE to + * ensure that upon release, propagation continues. + * Additionally, we must loop in case a new node is added + * while we are doing this. Also, unlike other uses of + * unparkSuccessor, we need to know if CAS to reset status + * fails, if so rechecking. + */ + for (;;) { + Node h = head; + if (h != null && h != tail) { + int ws = h.waitStatus; + if (ws == Node.SIGNAL) { + if (!compareAndSetWaitStatus(h, Node.SIGNAL, 0)) + continue; // loop to recheck cases + unparkSuccessor(h); + } + else if (ws == 0 && + !compareAndSetWaitStatus(h, 0, Node.PROPAGATE)) + continue; // loop on failed CAS + } + if (h == head) // loop if head changed + break; + } + } + + /** + * Sets head of queue, and checks if successor may be waiting + * in shared mode, if so propagating if either propagate > 0 or + * PROPAGATE status was set. + * + * @param node the node + * @param propagate the return value from a tryAcquireShared + */ + private void setHeadAndPropagate(Node node, int propagate) { + Node h = head; // Record old head for check below + setHead(node); + /* + * Try to signal next queued node if: + * Propagation was indicated by caller, + * or was recorded (as h.waitStatus either before + * or after setHead) by a previous operation + * (note: this uses sign-check of waitStatus because + * PROPAGATE status may transition to SIGNAL.) + * and + * The next node is waiting in shared mode, + * or we don't know, because it appears null + * + * The conservatism in both of these checks may cause + * unnecessary wake-ups, but only when there are multiple + * racing acquires/releases, so most need signals now or soon + * anyway. + */ + if (propagate > 0 || h == null || h.waitStatus < 0 || + (h = head) == null || h.waitStatus < 0) { + Node s = node.next; + if (s == null || s.isShared()) + doReleaseShared(); + } + } + + // Utilities for various versions of acquire + + /** + * Cancels an ongoing attempt to acquire. + * + * @param node the node + */ + private void cancelAcquire(Node node) { + // Ignore if node doesn't exist + if (node == null) + return; + + node.thread = null; + + // Skip cancelled predecessors + Node pred = node.prev; + while (pred.waitStatus > 0) + node.prev = pred = pred.prev; + + // predNext is the apparent node to unsplice. CASes below will + // fail if not, in which case, we lost race vs another cancel + // or signal, so no further action is necessary. + Node predNext = pred.next; + + // Can use unconditional write instead of CAS here. + // After this atomic step, other Nodes can skip past us. + // Before, we are free of interference from other threads. + node.waitStatus = Node.CANCELLED; + + // If we are the tail, remove ourselves. + if (node == tail && compareAndSetTail(node, pred)) { + compareAndSetNext(pred, predNext, null); + } else { + // If successor needs signal, try to set pred's next-link + // so it will get one. Otherwise wake it up to propagate. + int ws; + if (pred != head && + ((ws = pred.waitStatus) == Node.SIGNAL || + (ws <= 0 && compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && + pred.thread != null) { + Node next = node.next; + if (next != null && next.waitStatus <= 0) + compareAndSetNext(pred, predNext, next); + } else { + unparkSuccessor(node); + } + + node.next = node; // help GC + } + } + + /** + * 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 {@code true} if thread should block + */ + private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { + int ws = pred.waitStatus; + if (ws == Node.SIGNAL) + /* + * This node has already set status asking a release + * to signal it, so it can safely park. + */ + return true; + if (ws > 0) { + /* + * Predecessor was cancelled. Skip over predecessors and + * indicate retry. + */ + do { + node.prev = pred = pred.prev; + } while (pred.waitStatus > 0); + pred.next = node; + } else { + /* + * waitStatus must be 0 or PROPAGATE. 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, ws, Node.SIGNAL); + } + return false; + } + + /** + * Convenience method to interrupt current thread. + */ + static void selfInterrupt() { + Thread.currentThread().interrupt(); + } + + /** + * Convenience method to park and then check if interrupted + * + * @return {@code true} if interrupted + */ + private final boolean parkAndCheckInterrupt() { + LockSupport.park(this); + 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. + */ + + /** + * Acquires 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 {@code true} if interrupted while waiting + */ + final boolean acquireQueued(final Node node, int arg) { + boolean failed = true; + try { + boolean interrupted = false; + for (;;) { + final Node p = node.predecessor(); + if (p == head && tryAcquire(arg)) { + setHead(node); + p.next = null; // help GC + failed = false; + return interrupted; + } + if (shouldParkAfterFailedAcquire(p, node) && + parkAndCheckInterrupt()) + interrupted = true; + } + } finally { + if (failed) + cancelAcquire(node); + } + } + + /** + * Acquires in exclusive interruptible mode. + * @param arg the acquire argument + */ + private void doAcquireInterruptibly(int arg) + throws InterruptedException { + final Node node = addWaiter(Node.EXCLUSIVE); + boolean failed = true; + try { + for (;;) { + final Node p = node.predecessor(); + if (p == head && tryAcquire(arg)) { + setHead(node); + p.next = null; // help GC + failed = false; + return; + } + if (shouldParkAfterFailedAcquire(p, node) && + parkAndCheckInterrupt()) + throw new InterruptedException(); + } + } finally { + if (failed) + cancelAcquire(node); + } + } + + /** + * Acquires in exclusive timed mode. + * + * @param arg the acquire argument + * @param nanosTimeout max wait time + * @return {@code true} if acquired + */ + private boolean doAcquireNanos(int arg, long nanosTimeout) + throws InterruptedException { + if (nanosTimeout <= 0L) + return false; + final long deadline = System.nanoTime() + nanosTimeout; + final Node node = addWaiter(Node.EXCLUSIVE); + boolean failed = true; + try { + for (;;) { + final Node p = node.predecessor(); + if (p == head && tryAcquire(arg)) { + setHead(node); + p.next = null; // help GC + failed = false; + return true; + } + nanosTimeout = deadline - System.nanoTime(); + if (nanosTimeout <= 0L) + return false; + if (shouldParkAfterFailedAcquire(p, node) && + nanosTimeout > spinForTimeoutThreshold) + LockSupport.parkNanos(this, nanosTimeout); + if (Thread.interrupted()) + throw new InterruptedException(); + } + } finally { + if (failed) + cancelAcquire(node); + } + } + + /** + * Acquires in shared uninterruptible mode. + * @param arg the acquire argument + */ + private void doAcquireShared(int arg) { + final Node node = addWaiter(Node.SHARED); + boolean failed = true; + 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(); + failed = false; + return; + } + } + if (shouldParkAfterFailedAcquire(p, node) && + parkAndCheckInterrupt()) + interrupted = true; + } + } finally { + if (failed) + cancelAcquire(node); + } + } + + /** + * Acquires in shared interruptible mode. + * @param arg the acquire argument + */ + private void doAcquireSharedInterruptibly(int arg) + throws InterruptedException { + final Node node = addWaiter(Node.SHARED); + boolean failed = true; + 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 + failed = false; + return; + } + } + if (shouldParkAfterFailedAcquire(p, node) && + parkAndCheckInterrupt()) + throw new InterruptedException(); + } + } finally { + if (failed) + cancelAcquire(node); + } + } + + /** + * Acquires in shared timed mode. + * + * @param arg the acquire argument + * @param nanosTimeout max wait time + * @return {@code true} if acquired + */ + private boolean doAcquireSharedNanos(int arg, long nanosTimeout) + throws InterruptedException { + if (nanosTimeout <= 0L) + return false; + final long deadline = System.nanoTime() + nanosTimeout; + final Node node = addWaiter(Node.SHARED); + boolean failed = true; + 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 + failed = false; + return true; + } + } + nanosTimeout = deadline - System.nanoTime(); + if (nanosTimeout <= 0L) + return false; + if (shouldParkAfterFailedAcquire(p, node) && + nanosTimeout > spinForTimeoutThreshold) + LockSupport.parkNanos(this, nanosTimeout); + if (Thread.interrupted()) + throw new InterruptedException(); + } + } finally { + if (failed) + cancelAcquire(node); + } + } + + // Main exported methods + + + + /** + * 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 {@code true} if acquired; {@code 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 {@code true} if acquired; {@code 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)) { + doReleaseShared(); + 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 {@code true} return does not guarantee that any + * other thread will ever acquire. + * + *
In this implementation, this operation returns in + * constant time. + * + * @return {@code true} if there may be other threads waiting to acquire + */ + 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. + * + *
In this implementation, this operation returns in + * constant time. + * + * @return {@code true} if there has ever been contention + */ + public final boolean hasContended() { + return head != null; + } + + /** + * Returns the first (longest-waiting) thread in the queue, or + * {@code null} if no threads are currently queued. + * + *
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 + * {@code null} 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() { + /* + * The first node is normally head.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. We try this twice before + * resorting to traversal. + */ + Node h, s; + Thread st; + if (((h = head) != null && (s = h.next) != null && + s.prev == head && (st = s.thread) != null) || + ((h = head) != null && (s = h.next) != null && + s.prev == head && (st = s.thread) != null)) + 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. If not, we continue on, safely + * traversing from tail back to head to find first, + * guaranteeing termination. + */ + + Node t = tail; + Thread firstThread = null; + while (t != null && t != head) { + Thread tt = t.thread; + if (tt != null) + firstThread = tt; + t = t.prev; + } + return firstThread; + } + + /** + * Returns true if the given thread is currently queued. + * + *
This implementation traverses the queue to determine + * presence of the given thread. + * + * @param thread the thread + * @return {@code true} if the given thread is on the queue + * @throws NullPointerException if the thread is 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; + } + + /** + * Returns {@code true} if the apparent first queued thread, if one + * exists, is waiting in exclusive mode. If this method returns + * {@code true}, and the current thread is attempting to acquire in + * shared mode (that is, this method is invoked from {@link + * #tryAcquireShared}) then it is guaranteed that the current thread + * is not the first queued thread. Used only as a heuristic in + * ReentrantReadWriteLock. + */ + final boolean apparentlyFirstQueuedIsExclusive() { + Node h, s; + return (h = head) != null && + (s = h.next) != null && + !s.isShared() && + s.thread != null; + } + + /** + * Queries whether any threads have been waiting to acquire longer + * than the current thread. + * + *
An invocation of this method is equivalent to (but may be + * more efficient than): + *
{@code + * getFirstQueuedThread() != Thread.currentThread() && + * hasQueuedThreads()}+ * + *
Note that because cancellations due to interrupts and + * timeouts may occur at any time, a {@code true} return does not + * guarantee that some other thread will acquire before the current + * thread. Likewise, it is possible for another thread to win a + * race to enqueue after this method has returned {@code false}, + * due to the queue being empty. + * + *
This method is designed to be used by a fair synchronizer to + * avoid barging. + * Such a synchronizer's {@link #tryAcquire} method should return + * {@code false}, and its {@link #tryAcquireShared} method should + * return a negative value, if this method returns {@code true} + * (unless this is a reentrant acquire). For example, the {@code + * tryAcquire} method for a fair, reentrant, exclusive mode + * synchronizer might look like this: + * + *
{@code + * protected boolean tryAcquire(int arg) { + * if (isHeldExclusively()) { + * // A reentrant acquire; increment hold count + * return true; + * } else if (hasQueuedPredecessors()) { + * return false; + * } else { + * // try to acquire normally + * } + * }}+ * + * @return {@code true} if there is a queued thread preceding the + * current thread, and {@code false} if the current thread + * is at the head of the queue or the queue is empty + * @since 1.7 + */ + public final boolean hasQueuedPredecessors() { + // The correctness of this depends on head being initialized + // before tail and on head.next being accurate if the current + // thread is first in queue. + Node t = tail; // Read fields in reverse initialization order + Node h = head; + Node s; + return h != t && + ((s = h.next) == null || s.thread != Thread.currentThread()); + } + + + // 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 to acquire + */ + 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
Acquires the lock if it is not held by another thread and returns + * immediately, setting the lock hold count to one. + * + *
If the current thread already holds the lock then the hold + * count is incremented by one and the method returns immediately. + * + *
If the lock is held by another thread then the + * current thread becomes disabled for thread scheduling + * purposes and lies dormant until the lock has been acquired, + * at which time the lock hold count is set to one. + */ + public void lock() { + sync.lock(); + } + + /** + * Acquires the lock unless the current thread is + * {@linkplain Thread#interrupt interrupted}. + * + *
Acquires the lock if it is not held by another thread and returns + * immediately, setting the lock hold count to one. + * + *
If the current thread already holds this lock then the hold count + * is incremented by one and the method returns immediately. + * + *
If the lock is held by another thread then the + * current thread becomes disabled for thread scheduling + * purposes and lies dormant until one of two things happens: + * + *
If the lock is acquired by the current thread then the lock hold + * count is set to one. + * + *
If the current thread: + * + *
In this implementation, as this method is an explicit + * interruption point, preference is given to responding to the + * interrupt over normal or reentrant acquisition of the lock. + * + * @throws InterruptedException if the current thread is interrupted + */ + public void lockInterruptibly() throws InterruptedException { + sync.acquireInterruptibly(1); + } + + /** + * Acquires the lock only if it is not held by another thread at the time + * of invocation. + * + *
Acquires the lock if it is not held by another thread and + * returns immediately with the value {@code true}, setting the + * lock hold count to one. Even when this lock has been set to use a + * fair ordering policy, a call to {@code tryLock()} will + * immediately acquire the lock if it is available, whether or not + * other threads are currently waiting for the lock. + * This "barging" behavior can be useful in certain + * circumstances, even though it breaks fairness. If you want to honor + * the fairness setting for this lock, then use + * {@link #tryLock(long, TimeUnit) tryLock(0, TimeUnit.SECONDS) } + * which is almost equivalent (it also detects interruption). + * + *
If the current thread already holds this lock then the hold + * count is incremented by one and the method returns {@code true}. + * + *
If the lock is held by another thread then this method will return + * immediately with the value {@code false}. + * + * @return {@code true} if the lock was free and was acquired by the + * current thread, or the lock was already held by the current + * thread; and {@code false} otherwise + */ + public boolean tryLock() { + return sync.nonfairTryAcquire(1); + } + + /** + * Acquires the lock if it is not held by another thread within the given + * waiting time and the current thread has not been + * {@linkplain Thread#interrupt interrupted}. + * + *
Acquires the lock if it is not held by another thread and returns + * immediately with the value {@code true}, setting the lock hold count + * to one. If this lock has been set to use a fair ordering policy then + * an available lock will not be acquired if any other threads + * are waiting for the lock. This is in contrast to the {@link #tryLock()} + * method. If you want a timed {@code tryLock} that does permit barging on + * a fair lock then combine the timed and un-timed forms together: + * + *
{@code + * if (lock.tryLock() || + * lock.tryLock(timeout, unit)) { + * ... + * }}+ * + *
If the current thread + * already holds this lock then the hold count is incremented by one and + * the method returns {@code true}. + * + *
If the lock is held by another thread then the + * current thread becomes disabled for thread scheduling + * purposes and lies dormant until one of three things happens: + * + *
If the lock is acquired then the value {@code true} is returned and + * the lock hold count is set to one. + * + *
If the current thread: + * + *
If the specified waiting time elapses then the value {@code false} + * is returned. If the time is less than or equal to zero, the method + * will not wait at all. + * + *
In this implementation, as this method is an explicit + * interruption point, preference is given to responding to the + * interrupt over normal or reentrant acquisition of the lock, and + * over reporting the elapse of the waiting time. + * + * @param timeout the time to wait for the lock + * @param unit the time unit of the timeout argument + * @return {@code true} if the lock was free and was acquired by the + * current thread, or the lock was already held by the current + * thread; and {@code false} if the waiting time elapsed before + * the lock could be acquired + * @throws InterruptedException if the current thread is interrupted + * @throws NullPointerException if the time unit is null + */ + public boolean tryLock(long timeout, TimeUnit unit) + throws InterruptedException { + return sync.tryAcquireNanos(1, unit.toNanos(timeout)); + } + + /** + * Attempts to release this lock. + * + *
If the current thread is the holder of this lock then the hold + * count is decremented. If the hold count is now zero then the lock + * is released. If the current thread is not the holder of this + * lock then {@link IllegalMonitorStateException} is thrown. + * + * @throws IllegalMonitorStateException if the current thread does not + * hold this lock + */ + public void unlock() { + sync.release(1); + } + + /** + * Returns a {@link Condition} instance for use with this + * {@link Lock} instance. + * + *
The returned {@link Condition} instance supports the same + * usages as do the {@link Object} monitor methods ({@link + * Object#wait() wait}, {@link Object#notify notify}, and {@link + * Object#notifyAll notifyAll}) when used with the built-in + * monitor lock. + * + *
A thread has a hold on a lock for each lock action that is not + * matched by an unlock action. + * + *
The hold count information is typically only used for testing and + * debugging purposes. For example, if a certain section of code should + * not be entered with the lock already held then we can assert that + * fact: + * + *
{@code + * class X { + * ReentrantLock lock = new ReentrantLock(); + * // ... + * public void m() { + * assert lock.getHoldCount() == 0; + * lock.lock(); + * try { + * // ... method body + * } finally { + * lock.unlock(); + * } + * } + * }}+ * + * @return the number of holds on this lock by the current thread, + * or zero if this lock is not held by the current thread + */ + public int getHoldCount() { + return sync.getHoldCount(); + } + + /** + * Queries if this lock is held by the current thread. + * + *
Analogous to the {@link Thread#holdsLock(Object)} method for + * built-in monitor locks, this method is typically used for + * debugging and testing. For example, a method that should only be + * called while a lock is held can assert that this is the case: + * + *
{@code + * class X { + * ReentrantLock lock = new ReentrantLock(); + * // ... + * + * public void m() { + * assert lock.isHeldByCurrentThread(); + * // ... method body + * } + * }}+ * + *
It can also be used to ensure that a reentrant lock is used + * in a non-reentrant manner, for example: + * + *
{@code + * class X { + * ReentrantLock lock = new ReentrantLock(); + * // ... + * + * public void m() { + * assert !lock.isHeldByCurrentThread(); + * lock.lock(); + * try { + * // ... method body + * } finally { + * lock.unlock(); + * } + * } + * }}+ * + * @return {@code true} if current thread holds this lock and + * {@code false} otherwise + */ + public boolean isHeldByCurrentThread() { + return sync.isHeldExclusively(); + } + + /** + * Queries if this lock is held by any thread. This method is + * designed for use in monitoring of the system state, + * not for synchronization control. + * + * @return {@code true} if any thread holds this lock and + * {@code false} otherwise + */ + public boolean isLocked() { + return sync.isLocked(); + } + + /** + * Returns {@code true} if this lock has fairness set true. + * + * @return {@code true} if this lock has fairness set true + */ + public final boolean isFair() { + return sync instanceof FairSync; + } + + /** + * Returns the thread that currently owns this lock, or + * {@code null} if not owned. When this method is called by a + * thread that is not the owner, the return value reflects a + * best-effort approximation of current lock status. For example, + * the owner may be momentarily {@code null} even if there are + * threads trying to acquire the lock but have not yet done so. + * This method is designed to facilitate construction of + * subclasses that provide more extensive lock monitoring + * facilities. + * + * @return the owner, or {@code null} if not owned + */ + protected Thread getOwner() { + return sync.getOwner(); + } + + /** + * Queries whether any threads are waiting to acquire this lock. Note that + * because cancellations may occur at any time, a {@code true} + * return does not guarantee that any other thread will ever + * acquire this lock. This method is designed primarily for use in + * monitoring of the system state. + * + * @return {@code true} if there may be other threads waiting to + * acquire the lock + */ + public final boolean hasQueuedThreads() { + return sync.hasQueuedThreads(); + } + + /** + * Queries whether the given thread is waiting to acquire this + * lock. Note that because cancellations may occur at any time, a + * {@code true} return does not guarantee that this thread + * will ever acquire this lock. This method is designed primarily for use + * in monitoring of the system state. + * + * @param thread the thread + * @return {@code true} if the given thread is queued waiting for this lock + * @throws NullPointerException if the thread is null + */ + public final boolean hasQueuedThread(Thread thread) { + return sync.isQueued(thread); + } + + /** + * Returns an estimate of the number of threads waiting to + * acquire this lock. 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 of the system state, not for synchronization + * control. + * + * @return the estimated number of threads waiting for this lock + */ + public final int getQueueLength() { + return sync.getQueueLength(); + } + + /** + * Returns a collection containing threads that may be waiting to + * acquire this lock. 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 + */ + protected Collection