Condition

使用

获取锁后对应sync的wait & notify（notifyAll）

await

// await 方法是获取了锁的状态下，所以不需要进行大量的cas操作
public final void await() throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    // 添加condition内部链表的节点
    Node node = addConditionWaiter();
    // 解放掉在sync queue中的节点
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    // 判断当前的线程是否在Sync队列中,判断节点是不是在同步队列中，
    // 因为上一步已经释放了锁，也就是说此时可能有线程已经获取锁同时可能已经调用了singal方法，
    // 如果已经唤醒，那么就不应该park了，而是退出while方法，从而继续争抢锁
    while (!isOnSyncQueue(node)) {
        LockSupport.park(this);
        if ((interruptMode = checkInterruptWhileWaiting(node)) != 0)
            break;
    }
    // acquire方法中的判断能否执行，并设置头结点等等
    // 此为condition在await处被唤醒后
    // 此处被唤醒的时候，节点已经处于sync q之中了
    if (acquireQueued(node, savedState) && interruptMode != THROW_IE)
        interruptMode = REINTERRUPT;
    if (node.nextWaiter != null) // clean up if cancelled
        unlinkCancelledWaiters();
    if (interruptMode != 0)
        reportInterruptAfterWait(interruptMode);
}

/**
 * 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(Thread thread, int waitStatus) { // Used by Condition
    //    this.waitStatus = waitStatus;
    //    this.thread = thread;
    //}
    Node node = new Node(Thread.currentThread(), Node.CONDITION);
    // 清除所有多余节点后，t==null 意味着头结点为空
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}

/**
 * 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.
 */
 // 移除当前节点 ，如果节点的waitStatus != Node.condition
 // 因为节点只有向后指正，所以只能遍历
 // Called only while holding lock. 关键
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;
    }
}

/**
 * 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 {
        // 当前线程锁了多少个status（重入了N次）
        int savedState = getState();
        // 从sync q中移除，调用的就是unlock的release方法
        if (release(savedState)) {
            failed = false;
            return savedState;
        } else {
            throw new IllegalMonitorStateException();
        }
    } finally {
        if (failed)
            // 一些错误的操作,导致节点被取消
            node.waitStatus = Node.CANCELLED;
    }
}

/**
 * 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) {
    // node.waitStatus == CONDITION 的时候说明还不子啊sync q中
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
        // 在add waiter 中的代码，t.nextWaiter = node，只有在sync q中会设置next，而condition中不会
    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);
}

signal

/**
 * 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)
        // 唤醒头结点（condition的）
        doSignal(first);
}

/**
 * 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;
        // 遍历整个队列，唤醒一个可以唤醒的node
        // 而signal则只有first != null的判断
    } while (!transferForSignal(first) &&
             (first = firstWaiter) != null);
}

/**
 * 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;
    // 正常情况下 unpark是不会在这里执行的
    // 和sync一样，notify不是实时生效，而是等sync退出后，通过unlock来唤醒后续节点
    if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL))
        LockSupport.unpark(node.thread);
    return true;
}