CountDownLatch的await和countDown方法簡單分析

await

調用sync.acquireSharedInterruptibly

public void await() throws InterruptedException {
    sync.acquireSharedInterruptibly(1);
}

sync.acquireSharedInterruptibly
調用tryAcquireShared方法返回<0執行doAcquireSharedInterruptibly

public final void acquireSharedInterruptibly(int arg) throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0)
        doAcquireSharedInterruptibly(arg);
}

tryAcquireShared
嘗試獲取共享鎖,獲取成功返回1，不然-1

protected int tryAcquireShared(int acquires) {
    return (getState() == 0) ? 1 : -1;
}

doAcquireSharedInterruptibly

private void doAcquireSharedInterruptibly(int arg)throws InterruptedException {
    final Node node = addWaiter(Node.SHARED);
    boolean failed = true;
    try {
        for (;;) {
            final Node p = node.predecessor();
            //若是前一個node爲隊頭，則經過tryAcquireShared嘗試獲取共享鎖
            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);
    }
}

addWaiter
調用addWaiter方法把隊尾設置爲當前node；若是隊尾爲空或者設置失敗則調用enq方法

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;
}

enq
調用enq方法隊尾爲空則建立空的隊尾和隊頭，不然從新設置隊尾爲當前node，設置成功返回。enq和addWaiter方法不一樣在於enq循環執行必定會執行成功，不存在失敗狀況

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;
            }
        }
    }
}

predecessor
調用predecessor方法獲取前一個node

final Node predecessor() throws NullPointerException {
    Node p = prev;
    if (p == null)
        throw new NullPointerException();
    else
        return p;
}

static final int CANCELLED = 1; //取消 
static final int SIGNAL = -1; //下個節點須要被喚醒 
static final int CONDITION = -2; //線程在等待條件觸發
static final int PROPAGATE = -3; //（共享鎖）狀態須要向後傳播

shouldParkAfterFailedAcquire
獲取當前node的前一個note的線程等待狀態，若是爲SIGNAL,那麼返回true,大於0經過循環將當前節點以前全部取消狀態的節點移出隊列;其餘狀時,利用compareAndSetWaitStatus使前節點的狀態爲-1;若是是第一次await時ws狀態是0，屢次await時ws狀態是0，最後確定返回true

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
    int ws = pred.waitStatus;
    if (ws == Node.SIGNAL)
        return true;
    if (ws > 0) {
        do {
            node.prev = pred = pred.prev;
        } while (pred.waitStatus > 0);
        pred.next = node;
    } else {
        compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
    }
    return false;
}

parkAndCheckInterrupt
調用park並返回線程是否已經中斷

private final boolean parkAndCheckInterrupt() {
    LockSupport.park(this);
    return Thread.interrupted();
}

park
調用UNSAFE.park阻塞當前線程

public static void park(Object blocker) {
    Thread t = Thread.currentThread();
    setBlocker(t, blocker);
    UNSAFE.park(false, 0L);
    setBlocker(t, null);
}

setBlocker
在當前線程t的parkBlockerOffset位置設置blocker的引用

private static void setBlocker(Thread t, Object arg) {
    // Even though volatile, hotspot doesn't need a write barrier here.
    UNSAFE.putObject(t, parkBlockerOffset, arg);
}

UNSAFE.park

/**
 * 阻塞一個線程直到<a href="#unpark"><code>unpark</code></a>出現、線程
 * 被中斷或者timeout時間到期。若是一個<code>unpark</code>調用已經出現了，
 * 這裏只計數。timeout爲0表示永不過時.當<code>isAbsolute</code>爲true時，
 * timeout是相對於新紀元以後的毫秒。不然這個值就是超時前的納秒數。這個方法執行時
 * 也可能不合理地返回(沒有具體緣由)
 * 
 * @param isAbsolute true if the timeout is specified in milliseconds from
 *                   the epoch.
 *                   若是爲true timeout的值是一個相對於新紀元以後的毫秒數
 * @param time either the number of nanoseconds to wait, or a time in
 *             milliseconds from the epoch to wait for.
 *             能夠是一個要等待的納秒數，或者是一個相對於新紀元以後的毫秒數直到
 *             到達這個時間點
 */
UNSAFE.park(false, 0L);

countDown

調用sync.releaseShared

public void countDown() {
    sync.releaseShared(1);
}

releaseShared
執行tryReleaseShared成功後執行doReleaseShared

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) {
        doReleaseShared();
        return true;
    }
    return false;
}

tryReleaseShared
更新state值爲state-1，若是state新值爲0返回true，不然false

protected boolean tryReleaseShared(int releases) {
    // Decrement count; signal when transition to zero
    for (;;) {
        int c = getState();
        if (c == 0)
            return false;
        int nextc = c-1;
        if (compareAndSetState(c, nextc))
            return nextc == 0;
    }
}

doReleaseShared
只要等待隊列有數據，獲取隊頭等待狀態，隊頭狀態=-1其餘node爲等待時，則把隊頭等待狀態置爲初始，且調用unparkSuccessor方法；隊頭狀態=0時，把隊頭狀態置爲-3傳播到下一node

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;
    }
}

unparkSuccessor
上面調用unparkSuccessor時，node的狀態已經更改成0，且node.next存在，執行unpark方法

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);
}

unpark
unpark執行完以後是如何更改head的?

public static void unpark(Thread thread) {
    if (thread != null)
        UNSAFE.unpark(thread);
}

UNSAFE.unpark

/**
 * Releases the block on a thread created by 
 * <a href="#park"><code>park</code></a>.  This method can also be used
 * to terminate a blockage caused by a prior call to <code>park</code>.
 * This operation is unsafe, as the thread must be guaranteed to be
 * live.  This is true of Java, but not native code.
 * 釋放被<a href="#park"><code>park</code></a>建立的在一個線程上的阻塞.這個
 * 方法也能夠被使用來終止一個先前調用<code>park</code>致使的阻塞.
 * 這個操做操做時不安全的,所以線程必須保證是活的.這是java代碼不是native代碼。
 * @param thread the thread to unblock.
 *           要解除阻塞的線程
 */
UNSAFE.unpark(thread);