【JUC】1、Java AbstractQueuedSynchronizer - 源码翻译部分（三）
本文是学习AQS期间记录的，一篇文章发太多内容，编辑的时候很卡，所以独立开一篇AQS篇：https://blog.csdn.net/hhy107107/article/details/108041627说明：在阅读AQS源码的时候顺便翻译的，只翻译了部分，大部分用的google翻译，不太通顺的改成了自己的理解。一般还过得去的（基本能看懂的），就没改。/** ORACLE PROPRIETARY/C
·
本文是学习AQS期间记录的，一篇文章发太多内容，编辑的时候很卡，所以独立开一篇
AQS篇：https://blog.csdn.net/hhy107107/article/details/108041627
说明：在阅读AQS源码的时候顺便翻译的，只翻译了部分，大部分用的google翻译，不太通顺的改成了自己的理解。一般还过得去的（基本能看懂的），就没改。
/*
* ORACLE PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
*/
/*
* Written by Doug Lea with assistance from members of JCP JSR-166
* Expert Group and released to the public domain, as explained at
* http://creativecommons.org/publicdomain/zero/1.0/
*/
package java.util.concurrent.locks;
import java.util.concurrent.TimeUnit;
import java.util.ArrayList;
import java.util.Collection;
import java.util.Date;
import sun.misc.Unsafe;
/**
* 提供一个框架，用于实现依赖于先进先出（FIFO）等待队列的阻塞锁和相关的同步器（信号量，事件等）。
* 此类旨在为大多数依赖单个原子{@code int}值表示状态的同步器提供有用的基础。子类必须定义更改此状态的受保护方法，
* 并定义该状态对于获取或释放此对象而言意味着什么。鉴于这些，此类中的其他方法将执行所有排队和阻塞机制。
* 子类可以维护其他状态字段，但是仅跟踪关于同步的使用方法{@link #getState}，{@link #setState}和{@link #compareAndSetState}操作的原子更新的{@code int}值。
* 此类支持默认互斥模式和共享模式之一或两者。 当以独占方式进行获取时，其他线程尝试进行的获取将无法成功。 由多个线程获取的共享模式可能（但不一定）成功。
* 该类不“理解”这些差异，只是从机械意义上说，当共享模式获取成功时，下一个等待线程（如果存在）还必须确定它是否也可以获取。
* 在不同模式下等待的线程共享相同的FIFO队列。 通常，实现子类仅支持这些模式之一，但例如可以在{@link ReadWriteLock}中发挥作用。
* 仅支持互斥模式或仅共享模式的子类无需定义支持未使用模式的方法。
*
* 此类定义了一个嵌套的{@link AbstractQueuedSynchronizer.ConditionObject}类，该类可以被支持独占模式的子类用作{@link Condition}实现，
* 为此，独占模式{@link #isHeldExclusively}报告是否相对于当前独占同步。 线程，使用当前{@link #getState}值调用的方法{@link #release}会完全释放此对象，
* 并且给定已保存的状态值，{@link #acquire}最终会将其恢复为先前的获取状态。 否则，没有{@code AbstractQueuedSynchronizer}方法会创建这样的条件，
* 因此，如果无法满足此约束，请不要使用它。 {@link AbstractQueuedSynchronizer.ConditionObject}的行为当然取决于其同步器实现的语义。
*
* 此类提供了内部队列的检查，检测和监视方法，以及条件对象的类似方法。 可以根据需要使用{@code AbstractQueuedSynchronizer}将它们导出到类中以实现其同步机制。
*
* 此类的序列化仅存储基本的原子整数维护状态，因此反序列化的对象具有空线程队列。 要求可序列化的典型子类将定义一个{@code readObject}方法，
* 该方法可在反序列化时将其恢复为已知的初始状态。
*
* 用法
* 要将此类用作同步器的基础，请使用{@link #getState}，{@link #setState}和/或{@link #compareAndSetState}检查和/或修改同步状态，重新定义以下方法（如适用）：
*
* <ul>
* <li> {@link #tryAcquire}
* <li> {@link #tryRelease}
* <li> {@link #tryAcquireShared}
* <li> {@link #tryReleaseShared}
* <li> {@link #isHeldExclusively}
* </ul>
*
* 默认情况下，这些方法中的每一个都会引发{@link UnsupportedOperationException}。 这些方法的实现必须在内部是线程安全的，并且通常应简短且不阻塞。
* 定义这些方法是使用此类的唯一受支持的方法。 所有其他方法都声明为{@code final}，因为它们不能独立变化。
*
* 您可能还会发现{@link AbstractOwnableSynchronizer}的继承方法对于跟踪拥有独占同步器的线程很有用。
* 鼓励您使用它们-这将启用监视和诊断工具，以帮助用户确定哪些线程持有锁。
*
* 即使此类基于内部FIFO队列，它也不会自动执行FIFO获取策略。 独占同步的核心采用以下形式：
*
* <pre>
* Acquire:
* while (!tryAcquire(arg)) {
* 使线程排队（如果尚未排队）
* 可能阻塞当前线程
* }
*
* Release:
* if (tryRelease(arg))
* 取消阻塞第一个排队的线程
* </pre>
*
* (共享模式相似，但可能涉及级联信号。)
*
* 因为获取队列中的获取检查是在排队之前被调用的，所以新获取线程可能会在被阻塞和排队的其他线程之前插入。
* 但是，如果需要，您可以定义{@code tryAcquire}和/或{@code tryAcquireShared}来通过内部调用一种或多种检查方法来禁用插入，从而提供公平的FIFO获取顺序。
* 特别是，如果{@link #hasQueuedPredecessors}（一种专门为公平同步器设计的方法）返回{@code true}，
* 则大多数公平同步器都可以定义{@code tryAcquire}以返回{@code false}。 其他变化也是可能的。
*
* 对于默认的冲突策略（也称greedy策略，renouncement策略或者convoy-avoidance），吞吐量和可伸缩性通常最高。
* 尽管不能保证这是公平的或避免饥饿，但允许较早排队的线程在较晚排队的线程之前进行重新竞争，并且每一次和入列的新线程重竞争时都有同样的机会胜出。
* 此外，虽然获取同步值不是通常意义上"自旋"的，但是在阻塞之前，它们可能会执行{@code tryAcquire}的多次调用，并插入其他计算。
* 如果仅短暂地保持排他同步，则这将带来自旋的大部分好处，而在不进行同步时，则不会带来很多负担。
* 如果需要，您可以通过在调用之前使用“快速路径”检查来获取方法来增强此功能，
* 可能会预先检查{@link #hasContended}和/或{@link #hasQueuedThreads}以仅在同步器可能不这样做的情况下这样做 争辩。
*
* 此类为实现同步提供了一个高效且易扩展的基础，部分是因为专门界定了同步器的使用范围，该同步器依赖于一个int值state，可以获取和释放；
* 部分是因为内部维护了一个FIFO等待队列；如果这些还不足以满足需求，
* 你能从更底层使用java.util.concurrent.atomic 原子包下的原子类、 自定义的队列java.util.Queue 以及LockSupport类提供阻塞支持来构建自定义的同步器。
*
* <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. While a non-reentrant lock
* does not strictly require recording of the current owner
* thread, this class does so anyway to make usage easier to monitor.
* It also supports conditions and exposes
* one of the instrumentation methods:
*
* <pre> {@code
* class Mutex implements Lock, java.io.Serializable {
*
* // Our internal helper class
* private static class Sync extends AbstractQueuedSynchronizer {
* // Reports whether in locked state
* protected boolean isHeldExclusively() {
* return getState() == 1;
* }
*
* // Acquires the lock if state is zero
* public boolean tryAcquire(int acquires) {
* assert acquires == 1; // Otherwise unused
* if (compareAndSetState(0, 1)) {
* setExclusiveOwnerThread(Thread.currentThread());
* return true;
* }
* return false;
* }
*
* // Releases the lock by setting state to zero
* protected boolean tryRelease(int releases) {
* assert releases == 1; // Otherwise unused
* if (getState() == 0) throw new IllegalMonitorStateException();
* setExclusiveOwnerThread(null);
* setState(0);
* return true;
* }
*
* // Provides a Condition
* Condition newCondition() { return new ConditionObject(); }
*
* // Deserializes 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 java.util.concurrent.CountDownLatch CountDownLatch}
* except that it only requires a single {@code signal} to
* fire. Because a latch is non-exclusive, it uses the {@code shared}
* acquire and release methods.
*
* <pre> {@code
* 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
extends AbstractOwnableSynchronizer
implements java.io.Serializable {
private static final long serialVersionUID = 7373984972572414691L;
/**
* 创建一个新的{@code AbstractQueuedSynchronizer}实例，其初始同步状态为零。
*/
protected AbstractQueuedSynchronizer() { }
/**
* 等待队列节点类。
*
* 等待队列是“ CLH”（Craig，Landin和Hagersten）锁定队列的变体。 CLH锁通常用于自旋锁。
* 相反，我们将它们用于阻塞同步器，但是使用相同的基本策略，即在其节点的前身中保存有关线程的某些控制信息。 A
* 每个节点中的“状态”字段跟踪线程是否应阻塞。 节点的前任释放时会发出信号。 否则，队列的每个节点都将用作持有单个等待线程的特定通知样式的监视器。
* 虽然状态字段不控制是否授予线程锁等。 线程可能会尝试获取它是否在队列中的第一位。 但是先行并不能保证成功。 它只赋予了竞争的权利。 因此，当前发布的竞争线程可能需要重新等待。
*
* 要加入CLH锁，您可以自动将其作为新尾部拼接。 要出队，您只需设置头字段。
* <pre>
* +------+ prev +-----+ +-----+
* head | | <---- | | <---- | | tail
* +------+ +-----+ +-----+
* </pre>
*
* 插入到CLH队列中，只需要对“尾节点”执行一次原子操作，因此存在一个简单的原子分界点，即从未排队到排队。
* 同样，出队仅涉及更新“头节点”。 但是，节点需要花费更多的精力来确定其后继者是谁，部分原因是要处理由于超时和中断而可能导致的取消。
*
* "prev"链接（在原始CLH锁中不使用）主要用于处理取消。如果取消某个节点，则其后继节点（通常）会重新链接到未取消的前任节点。
* 有关自旋锁情况下类似机制的说明，请参见Scott和Scherer的论文，网址为http://www.cs.rochester.edu/u/scott/synchronization/
*
* 我们还使用“next”链接来实现阻止机制。每个节点的线程ID保留在其自己的节点中，因此，
* 前任通过遍历下一个链接以确定它是哪个线程，来通知下一个节点唤醒。
* 确定后继者必须避免与新排队的节点竞争以设置其前任节点的“ next”字段。
* 在必要时，可以通过在节点的后继者似乎为空时从原子更新的“尾部”向后检查来解决此问题。
* （或者换句话说，下一个链接是一种优化，因此我们通常不需要向后扫描。）
*
* <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, unless we can identify an uncancelled
* predecessor who will carry this responsibility.
*
* CLH队列需要一个虚拟标头节点才能开始。 但是我们不会在构建过程中创建它们，因为如果没有争用，这将是浪费的精力。 而是构造节点，并在第一次争用时设置头和尾指针。
*
* 等待条件的线程使用相同的节点，但使用附加链接。 条件只需要在简单（非并发）链接队列中链接节点，因为只有在专用时才可以访问它们。
* 等待时，将节点插入条件队列。 收到信号后，该节点将转移到主队列。 状态字段的特殊值用于标记节点所在的队列。
*
*/
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;
/** 当前线程在等待condition */
static final int CONDITION = -2;
/**
* 表示后续的acquireShared都能得到执行
*/
static final int PROPAGATE = -3;
/**
* 状态字段，仅采用以下值：
* SIGNAL: 后继节点线程被阻塞或即将阻塞(通过park), 所以当前节点被释放或取消的时候，必须唤醒（unpark）后继节点线程。
* 为了避免冲突，qcquire 方法必须先的的状态是signal,然后再原子的获取，如果失败，阻塞。
* CANCELLED: 由于超时或中断，该节点被取消。 节点永远不会离开此状态。 特别是，具有取消节点的线程永远不会再次阻塞。
* CONDITION: 该节点当前在条件队列中。 在传输之前，它不会用作同步队列节点，此时状态将设置为0。（此值的使用与该字段的其他用途无关，但简化了机制。）
* PROPAGATE: releaseShared应该传播到其他节点。 在doReleaseShared中对此进行了设置（仅适用于头节点），以确保传播继续进行，即使此后进行了其他操作也是如此。
* 0: 以上都不是
*
* 这些值以数字方式排列以简化使用。 非负值表示节点不需要发信号。 因此，大多数代码不需要检查特定值，仅需检查符号即可。
*
* 对于常规同步节点，该字段初始化为0，对于条件节点，该字段初始化为CONDITION。 使用CAS（或在可能的情况下进行无条件的volatile写操作）对其进行修改。
*/
volatile int waitStatus;
/**
* 链接到当前节点/线程所依赖的先前节点，以检查waitStatus。 在入队期间分配，并且仅在出队时将其清空（出于GC的考虑）。
* 同样，在取消前任后，我们会短路，同时找到一个未取消的前任，这将始终存在，因为根节点从未被取消：只有成功获取后，结点才变为根。
* 取消的线程永远不会成功获取，并且线程只会取消自身，不会取消任何其他节点。
*/
volatile Node prev;
/**
* 链接到后继节点，当前节点/线程在释放时将其解散。 在排队时分配，在绕过取消的前任时进行调整，在出队时清零（出于GC的考虑）。
* enq操作直到附加后才分配前任的下一个字段，因此看到空的下一个字段不一定表示节点在队列末尾。
* 但是，如果下一个字段似乎为空，则我们可以从尾部扫描上一个以进行再次检查。
* 被取消节点的下一个字段设置为指向节点本身而不是null，以使isOnSyncQueue的工作更轻松。
*/
volatile Node next;
/**
* 使该节点排队的线程。 在构造上初始化，使用后消失。
*/
volatile Thread thread;
/**
* 链接到等待条件的下一个节点，或者链接到特殊值SHARED。
* 由于条件队列仅在以独占模式保存时才被访问，因此我们只需要一个简单的链接队列即可在节点等待条件时保存节点。
* 然后将它们转移到队列中以重新获取。 由于条件只能是互斥的，因此我们使用特殊值来表示共享模式来保存字段。
*/
Node nextWaiter;
/**
* 如果节点在共享模式下等待，则返回true。
*/
final boolean isShared() {
return nextWaiter == SHARED;
}
/**
* 链接到等待条件的下一个节点，或者链接到特殊值SHARED。
* 由于条件队列仅在以独占模式保存时才被访问，因此我们只需要一个简单的链接队列即可在节点等待条件时保存节点。
* 然后将它们转移到队列中以重新获取。
* 由于条件只能是互斥的，因此我们使用特殊值来表示共享模式来保存字段，返回上一个节点，
* 如果为null则抛出NullPointerException。 当前任不能为null时使用。 空检查可能会被忽略，但是它可以帮助VM。
*
* @return 该节点的前驱
*/
final Node predecessor() throws NullPointerException {
Node p = prev;
if (p == null)
throw new NullPointerException();
else
return p;
}
Node() { // 用于建立初始标头或SHARED标记
}
Node(Thread thread, Node mode) { // 用于 addWaiter
this.nextWaiter = mode;
this.thread = thread;
}
Node(Thread thread, int waitStatus) { // 用于 Condition
this.waitStatus = waitStatus;
this.thread = thread;
}
}
/**
* 等待队列的头，延迟初始化。 除初始化外，只能通过setHead方法进行修改。 注意：如果head存在，则保证其waitStatus不被设置成CANCELLED。
*/
private transient volatile Node head;
/**
* 等待队列的尾部，延迟初始化。 仅通过方法enq进行修改以添加新的等待节点。
*/
private transient volatile Node tail;
/**
* 同步状态。
*/
private volatile int state;
/**
* 返回同步状态的当前值。 此操作具有{@code volatile}读取的内存语义。
* @return 当前state值
*/
protected final int getState() {
return state;
}
/**
* 设置同步状态的值。 此操作具有{@code volatile}写操作的内存语义。
* @param newState 设置state
*/
protected final void setState(int newState) {
state = newState;
}
/**
* 如果当前状态值等于期望值，则以原子方式将同步状态设置为给定的更新值。 该操作具有{@code volatile}读写的内存语义。
*
* @param expect 期望值
* @param update 新值
* @return {@code true} 成功. {@code false} 实际值不等于期望值。
*/
protected final boolean compareAndSetState(int expect, int update) {
// 请参阅下面的内部函数设置以支持此操作
return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}
// 排队工具
/**
* 为了防止自定义的时间限过长，而设置的，如果设置的时间限长于这个值则取这个spinForTimeoutThreshold 为时间限
* 粗略估计足以在非常短的超时时间内提高响应速度。
*/
static final long spinForTimeoutThreshold = 1000L;
/**
* 将节点插入队列，必要时进行初始化。 参见上图。
* @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
if (compareAndSetHead(new Node()))
tail = head;
} else {
node.prev = t;
if (compareAndSetTail(t, node)) {
t.next = node;
return t;
}
}
}
}
/**
* 为当前线程和给定模式创建并排队节点。
*
* @param mode Node.EXCLUSIVE用于独占，Node.SHARED用于共享
* @return 新节点
*/
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;
}
/**
* 将node设置成队列头. 仅通过acquire方法调用
*
* @param node the node
*/
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}
/**
* 唤醒节点的后继者（如果存在）。
*
* @param node the node
*/
private void unparkSuccessor(Node node) {
/*
* 如果当前节点的状态是负数，尝试修改自己的状态为0
* 只要状态被改了就行，不管是不是当前线程改的..
*/
int ws = node.waitStatus;
if (ws < 0)
compareAndSetWaitStatus(node, ws, 0);
/*
* unpark一个后继者线程（通常是节点的后继）。
* 如果后继为空或者状态>0, 那么从尾部开始遍历，找到一个需要唤醒（状态<0）的线程，做唤醒操作
*/
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);
}
/**
* 释放共享锁，并想后继发信号（确保事件传播）
* （注意：对于独占模式，如果需要发信号，释放仅相当于调用head的unparkSuccessor。）
*/
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;
}
}
/**
* 将node设置成头节点
* 如果propagate > 0, 或者node节点的状态为PROPAGATE，释放共享模式（向后传递）
*
* @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
/**
* 取消正在进行的acquire
*
* @param node the node
*/
private void cancelAcquire(Node node) {
// 节点不存在, 忽略
if (node == null)
return;
node.thread = null;
// 跳过已经取消了的前驱
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
}
}
/**
* 在acquire失败后，检查并更新节点状态
* 如果线程应该被阻塞(需要等待获取一个许可)，返回true. This is the main signal
* control in all acquire loops. Requires that pred == node.prev.
*
* @param pred 节点的前驱
* @param node 当前节点
* @return {@code true} if thread should block
*/
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL)
/*
* 节点已经设置了signal，需要唤醒
*/
return true;
if (ws > 0) {
/*
* 前驱节点已取消，跳过前驱，往继续往前找。
*/
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node;
} else {
/*
* 声明为我们需要的状态：SIGNAL
* 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();
}
/**
* 等待许可，并检查中断
*
* @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.
*/
/**
* 不间断地为node获取排他许可（node已经入队）
* 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;
}
// shouldParkAfterFailedAcquire：如果p.status = SIGNAL, 返回ture， 反之将status设置为SIGNAL，返回false
// parkAndCheckInterrupt 等待许可，并检查中断
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
/**
* 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 {@code 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 {@code true} if this object is now in a fully released
* state, so that any waiting threads may attempt to acquire;
* and {@code false} 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 if acquisition in shared
* mode succeeded but no subsequent shared-mode acquire can
* succeed; and a positive value if acquisition in shared
* mode succeeded and subsequent shared-mode acquires might
* also succeed, 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 {@code true} if this release of shared mode may permit a
* waiting acquire (shared or exclusive) to succeed; and
* {@code false} 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 {@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}.)
*
* <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 {@code true} if synchronization is held exclusively;
* {@code false} otherwise
* @throws UnsupportedOperationException if conditions are not supported
*/
protected boolean isHeldExclusively() {
throw new UnsupportedOperationException();
}
/**
* 以互斥方式获取，忽略中断。
* 通过调用{@link #tryAcquire}来实现，如果返回true,就是acquire成功；
* 否则，会一直调用{@link #tryAcquire}，直到成功，期间线程将排队，并可能反复阻塞和解除阻塞。
* 此方法可用于实现方法{@link Lock＃lock}。
* @param arg 参数会被传递到{@link #tryAcquire}方法中，可以用来做任何事
*/
public final void acquire(int arg) {
if (!tryAcquire(arg) &&
acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
selfInterrupt();
}
/**
* 以互斥方式获取，会被中断。
* 方法先检查中断状态
* 通过调用{@link #tryAcquire}来实现，如果返回true,就是acquire成功；
* 否则，会一直调用{@link #tryAcquire}，直到成功，期间线程将排队，并可能反复阻塞和解除阻塞。
* 此方法可用于实现方法{@link Lock#lockInterruptibly}。
* @param arg 参数会被传递到{@link #tryAcquire}方法中，可以用来做任何事
* @throws InterruptedException 如果线程中断，会抛出异常
*/
public final void acquireInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (!tryAcquire(arg))
doAcquireInterruptibly(arg);
}
/**
* 以互斥方式获取，会被中断。如果超过给定的时间，则失败
* 方法先检查中断状态
* 通过调用{@link #tryAcquire}来实现，如果返回true,就是acquire成功；
* 否则，会一直调用{@link #tryAcquire}，直到成功，期间线程将排队，并可能反复阻塞和解除阻塞。
* 此方法可用于实现方法{@link Lock＃lock}。
* @param arg 参数会被传递到{@link #tryAcquire}方法中，可以用来做任何事
* @param nanosTimeout 最大等待纳秒数
* @return {@code true} 获取成功; {@code false} 超时
* @throws InterruptedException 如果线程中断，会抛出异常
*/
public final boolean tryAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
return tryAcquire(arg) ||
doAcquireNanos(arg, nanosTimeout);
}
/**
* 释放互斥模式.
* Implemented by unblocking one or more threads if {@link #tryRelease} returns true.
* 此方法可用于实现方法{@link Lock#unlock}.
*
* @param arg 参数会被传递到{@link #tryRelease}方法中，可以用来做任何事
* @return 返回 {@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;
}
/**
* 以共享模式获取，忽略中断。
* 通过首先至少调用一次{@link #tryAcquireShared}并成功返回来实现。
* 否则，线程将排队，并可能反复阻塞和解除阻塞，并调用{@link #tryAcquireShared}直到成功。
*
* @param arg 参数会被传递到{@link #tryAcquireShared}方法中，可以用来做任何事
*/
public final void acquireShared(int arg) {
if (tryAcquireShared(arg) < 0)
doAcquireShared(arg);
}
/**
* 以共享模式获取，如果中断会被中止。
* 方法先判断线程是否被中断
* 通过首先至少调用一次{@link #tryAcquireShared}并成功返回来实现。
* 否则，线程将排队，并可能反复阻塞和解除阻塞，并调用{@link #tryAcquireShared}直到成功。
*
* @param arg 参数会被传递到{@link #tryAcquireShared}方法中，可以用来做任何事
* @throws InterruptedException 如果线程中断，会抛出异常
*/
public final void acquireSharedInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
if (tryAcquireShared(arg) < 0)
doAcquireSharedInterruptibly(arg);
}
/**
* 以共享模式获取，如果中断会被中止。如果超过超时时间，返回false
* 方法先判断线程是否被中断
* 通过首先至少调用一次{@link #tryAcquireShared}并成功返回来实现。
* 否则，线程将排队，并可能反复阻塞和解除阻塞，并调用{@link #tryAcquireShared}直到成功。
*
* @param arg 参数会被传递到{@link #tryAcquireShared}方法中，可以用来做任何事
* @param nanosTimeout 最大等待纳秒数
* @return {@code true} 获取成功; {@code false} 超时
* @throws InterruptedException 如果线程中断，会抛出异常
*/
public final boolean tryAcquireSharedNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
return tryAcquireShared(arg) >= 0 ||
doAcquireSharedNanos(arg, nanosTimeout);
}
/**
* 释放共享模式. Implemented by unblocking one or more
* threads if {@link #tryReleaseShared} returns true.
*
* @param arg 参数会被传递到{@link #tryReleaseShared}方法中，可以用来做任何事
* @return 返回 {@link #tryReleaseShared} 方法的返回值
*/
public final boolean releaseShared(int arg) {
if (tryReleaseShared(arg)) {
doReleaseShared();
return true;
}
return false;
}
// 队列检查方法
/**
* 查询是否有任何线程正在等待获取。
* 请注意，由于中断和超时引起的取消可能随时发生，因此{@code true}返回值不能保证任何其他线程都可以获取。
*
* 在此实现中，此操作将以固定的时间返回。
*
* @return {@code true} 可能还有其他线程在等待获取
*/
public final boolean hasQueuedThreads() {
return head != tail;
}
/**
* 查询是否有任何线程争夺过该同步器； 也就是说，如果true,表示有线程调用acquire但被阻止了。
*
* 在此实现中，此操作将以固定的时间返回。
*
* @return {@code true} 曾经有线程调用acquire但被阻止了。
*/
public final boolean hasContended() {
return head != null;
}
/**
* 返回队列中的第一个（等待时间最长）线程，如果当前没有线程在排队，则返回{@code null}。
*
* <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
* {@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() {
/*
* 第一个节点通常是head.next, 尝试获取head.next.thread
* 为确保一致读，判断两次
* 如果head不为null，并且head的后驱节点不为null，并且head的后驱节点的前驱节点是head节点
*/
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的下一个字段可能尚未设置，或者在setHead之后可能未设置。
* 因此，我们必须检查一下tail是否实际上是第一个节点。
* 如果没有，从tail往前找
*/
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;
}
/**
* 如果给定的线程在队列中，返回true
*
* 需要遍历队列，从tail开始
*
* @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;
}
/**
* 如果明显的第一个排队线程（如果存在）正以独占模式等待，则返回{@code true}。
* 如果此方法返回{@code true}，并且当前线程正试图以共享模式获取（即，此方法是从{@link#tryAcquireShared} ）调用的，
* 那么可以保证当前线程不是第一个排队的线程。只在可重入写锁中用作启发式。
*/
final boolean apparentlyFirstQueuedIsExclusive() {
Node h, s;
return (h = head) != null &&
(s = h.next) != null &&
!s.isShared() &&
s.thread != null;
}
/**
* 查询是否有任何线程在等待获取比当前线程更长的时间。
*
* 调用此方法等效于（但可能比以下方法更有效）：
* {@code getFirstQueuedThread() != Thread.currentThread() && hasQueuedThreads()}
*
* 请注意，由于中断和超时导致的取消可能随时发生，因此{@code true}返回值不能保证一定存在某些其他线程排在当前线程之前。
* 同样，由于队列为空，此方法返回{@code false}后，另一个线程也有可能获取到了锁。
*
* 此方法设计为由公平同步器使用，
* <p>This method is designed to be used by a fair synchronizer to
* avoid <a href="AbstractQueuedSynchronizer#barging">barging</a>.
*
* 如果这个方法返回true, 同步器的tryAcquire方法应该返回false, 并且tryAcquireShared方法应该返回一个负值（除非这个是可重入的）
* 例如，用于公平，可重入，互斥模式同步器的{@code tryAcquire}方法可能如下所示：:
*
* <pre> {@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
* }
* }}</pre>
*
* @return {@code true} 如果当前线程之前有一个排队的线程; {@code false} 如果当前线程位于队列的开头或队列为空
* @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<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 {@code "State ="}
* followed by the current value of {@link #getState}, and either
* {@code "nonempty"} or {@code "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 ws = p.waitStatus;
if (ws > 0 || !compareAndSetWaitStatus(p, ws, 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 node the 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;
}
/**
* Invokes release with current state value; returns saved state.
* Cancels node and throws exception on failure.
* @param node the condition node for this wait
* @return previous sync state
*/
final int fullyRelease(Node node) {
boolean failed = true;
try {
int savedState = getState();
if (release(savedState)) {
failed = false;
return savedState;
} else {
throw new IllegalMonitorStateException();
}
} finally {
if (failed)
node.waitStatus = Node.CANCELLED;
}
}
// Instrumentation methods for conditions
/**
* Queries whether the given ConditionObject
* uses this synchronizer as its lock.
*
* @param condition the condition
* @return {@code true} if owned
* @throws NullPointerException if the condition is null
*/
public final boolean owns(ConditionObject condition) {
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 {@code true} return
* does not guarantee that a future {@code signal} will awaken
* any threads. This method is designed primarily for use in
* monitoring of the system state.
*
* @param condition the condition
* @return {@code true} 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 the condition is 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 the condition is 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 the condition is 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
* {@code AbstractQueuedSynchronizer}.
*
* <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 {@code ConditionObject} instance.
*/
public ConditionObject() { }
// Internal methods
/**
* Adds a new waiter to wait queue.
* @return its new wait node
*/
private Node addConditionWaiter() {
Node t = lastWaiter;
// If lastWaiter is cancelled, clean out.
if (t != null && t.waitStatus != Node.CONDITION) {
unlinkCancelledWaiters();
t = lastWaiter;
}
Node node = new Node(Thread.currentThread(), Node.CONDITION);
if (t == null)
firstWaiter = node;
else
t.nextWaiter = node;
lastWaiter = node;
return node;
}
/**
* Removes and transfers 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);
}
/**
* Removes and transfers 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);
}
/**
* Unlinks cancelled waiter nodes from condition queue.
* Called only while holding lock. This is called when
* cancellation occurred during condition wait, and upon
* insertion of a new waiter when lastWaiter is seen to have
* been cancelled. This method is needed to avoid garbage
* retention in the absence of signals. So even though it may
* require a full traversal, it comes into play only when
* timeouts or cancellations occur in the absence of
* signals. It traverses all nodes rather than stopping at a
* particular target to unlink all pointers to garbage nodes
* without requiring many re-traversals during cancellation
* storms.
*/
private void unlinkCancelledWaiters() {
Node t = firstWaiter;
Node trail = null;
while (t != null) {
Node next = t.nextWaiter;
if (t.waitStatus != Node.CONDITION) {
t.nextWaiter = null;
if (trail == null)
firstWaiter = next;
else
trail.nextWaiter = next;
if (next == null)
lastWaiter = trail;
}
else
trail = t;
t = next;
}
}
// 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 {@code 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 {@code 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(this);
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;
/**
* Checks 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;
}
/**
* Throws InterruptedException, reinterrupts current thread, or
* does nothing, depending on mode.
*/
private void reportInterruptAfterWait(int interruptMode)
throws InterruptedException {
if (interruptMode == THROW_IE)
throw new InterruptedException();
else if (interruptMode == REINTERRUPT)
selfInterrupt();
}
/**
* 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 InterruptedException.
* </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(this);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null) // clean up if cancelled
unlinkCancelledWaiters();
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);
final long deadline = System.nanoTime() + nanosTimeout;
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
if (nanosTimeout <= 0L) {
transferAfterCancelledWait(node);
break;
}
if (nanosTimeout >= spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
nanosTimeout = deadline - System.nanoTime();
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
return deadline - System.nanoTime();
}
/**
* 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 {
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(this, abstime);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
unlinkCancelledWaiters();
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 {
long nanosTimeout = unit.toNanos(time);
if (Thread.interrupted())
throw new InterruptedException();
Node node = addConditionWaiter();
int savedState = fullyRelease(node);
final long deadline = System.nanoTime() + nanosTimeout;
boolean timedout = false;
int interruptMode = 0;
while (!isOnSyncQueue(node)) {
if (nanosTimeout <= 0L) {
timedout = transferAfterCancelledWait(node);
break;
}
if (nanosTimeout >= spinForTimeoutThreshold)
LockSupport.parkNanos(this, nanosTimeout);
if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
break;
nanosTimeout = deadline - System.nanoTime();
}
if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
interruptMode = REINTERRUPT;
if (node.nextWaiter != null)
unlinkCancelledWaiters();
if (interruptMode != 0)
reportInterruptAfterWait(interruptMode);
return !timedout;
}
// support for instrumentation
/**
* Returns true if this condition was created by the given
* synchronization object.
*
* @return {@code 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(ConditionObject)}.
*
* @return {@code true} if there are any waiting threads
* @throws IllegalMonitorStateException if {@link #isHeldExclusively}
* returns {@code 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(ConditionObject)}.
*
* @return the estimated number of waiting threads
* @throws IllegalMonitorStateException if {@link #isHeldExclusively}
* returns {@code 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(ConditionObject)}.
*
* @return the collection of threads
* @throws IllegalMonitorStateException if {@link #isHeldExclusively}
* returns {@code 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;
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); }
}
/**
* 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 static final boolean compareAndSetWaitStatus(Node node,
int expect,
int update) {
return unsafe.compareAndSwapInt(node, waitStatusOffset,
expect, update);
}
/**
* CAS next field of a node.
*/
private static final boolean compareAndSetNext(Node node,
Node expect,
Node update) {
return unsafe.compareAndSwapObject(node, nextOffset, expect, update);
}
}
开放原子开发者工作坊旨在鼓励更多人参与开源活动，与志同道合的开发者们相互交流开发经验、分享开发心得、获取前沿技术趋势。工作坊有多种形式的开发者活动，如meetup、训练营等，主打技术交流，干货满满，真诚地邀请各位开发者共同参与！
更多推荐
0
0- 0
已为社区贡献1条内容
所有评论(0)