juc并发包-lucks

 * 同步状态.
 */
private volatile int state;

protected final int () {
    return state;
}

protected final void setState(int newState) {
    state = newState;
}

 * 以原子的方式更新同步状态.
 * 利用Unsafe类实现
 */
protected final boolean compareAndSetState(int expect, int update) {
    return unsafe.compareAndSwapInt(this, stateOffset, expect, update);
}

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;

    
    * INITAL：      0 - 默认，新结点会处于这种状态。
    * CANCELLED：   1 - 取消，表示当前结点被中断或超时，需要移出队列；
    * SIGNAL：      -1- 发信号，表示后续结点被阻塞了；（当前结点在入队后、阻塞前，应确保将其prev结点类型改为SIGNAL，以便prev结点取消或释放时将当前结点唤醒。）
    * CONDITION：   -2- Condition专用，表示当前结点在Condition队列中，因为等待某个条件而被阻塞了；
    * PROPAGATE：   -3- 传播，适用于共享模式。（比如连续的读操作结点可以依次进入临界区，设为PROPAGATE有助于实现这种迭代操作。）
    * 
    * waitStatus表示的是后续结点状态，这是因为AQS中使用CLH队列实现线程的结构管理，而CLH结构正是用前一结点某一属性表示当前结点的状态，这样更容易实现取消和超时功能。
    */
    volatile int waitStatus;

    // 前驱指针
    volatile Node prev;

    // 后驱指针
    volatile Node next;

    // 结点所包装的线程
    volatile Thread thread;

    // Condition队列使用，存储condition队列中的后继节点
    Node nextWaiter;

    Node() {
    }

    Node(Thread thread, Node mode) { 
        this.nextWaiter = mode;
        this.thread = thread;
    }
}


 
     * 添加节点到等待队列, 如果有必须就初始化Node. 
    *  自选CAS添加等待队列尾部. 直到成功
     * @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;
                }
            }
        }
    }

    
     * Creates and enqueues node for current thread and given mode.
    * 创建node并入队, 设置当期线程和指定锁的类型 是独占还是共享
     *
     * @param mode Node.EXCLUSIVE for exclusive, Node.SHARED for shared
     * @return the new node
     */
    private Node addWaiter(Node mode) {
        Node node = new Node(Thread.currentThread(), mode);
        // Try the fast path of enq; backup to full enq on failure
        // 尝试快速设置 将node加入队列尾. 如果失败了就交个end方法. 自旋去
        Node pred = tail;
        if (pred != null) {
            node.prev = pred;
            if (compareAndSetTail(pred, node)) {
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

public void lock() {
    sync.lock();
}

 static final class FairSync extends Sync {
    private static final long serialVersionUID = -3000897897090466540L;

    final void lock() {
        acquire(1);
    }
}

 public final void acquire(int arg) { // arg =1
    if (
        !tryAcquire(arg) //尝试获取资源
        && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        selfInterrupt();
}

 
    * 
     * Fair version of tryAcquire.  Don't grant access unless
     * recursive call or no waiters or is first.
     */
    protected final boolean tryAcquire(int acquires) { //acquires =1
        final Thread current = Thread.currentThread(); //获得当期线程
        int c = getState(); // 获得同步状态
        if (c == 0) { // 0代表锁并未被占用
            //如果等待队列中. 在当期线程前没有其他线程. 则用CAS获得锁
            if (!hasQueuedPredecessors() &&
                compareAndSetState(0, acquires)) {
                //更新成功. 将锁的占有线程设置为当期线程
                setExclusiveOwnerThread(current);
                return true;
            }
        }
        //c!=0 说明锁被占用. 判断是否是重入的情况
        else if (current == getExclusiveOwnerThread()) {
            // state+1
            int nextc = c + acquires;
            // 重入次数过大. 溢出了
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
                //设置状态
            setState(nextc);
            return true;
        }
        return false;
    }

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

public final void acquire(int arg) { // arg =1
    if (
        //尝试获取锁 这里是失败的. 因为A已经获得锁了. c!=0 并且 current != getExclusiveOwnerThread()
        !tryAcquire(arg) //尝试获取资源
        //将当期线程包装成节点加入等待队列
        && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
        // 中断自己
        selfInterrupt();
}

private Node addWaiter(Node mode) {
    // 将线程包装成节点
    Node node = new Node(Thread.currentThread(), mode);
    // Try the fast path of enq; backup to full enq on failure
    // 视图快速将自己设置成尾节点. 先尝试一次, 如果成功就不用end() 失败. 自旋去 tail!=null 说明队列中有排队的node. 那么让当期node排在他后面. 把自己改为tail即可 
    Node pred = tail;
    if (pred != null) {
        node.prev = pred;
        if (compareAndSetTail(pred, node)) {
            pred.next = node;
            return node;
        }
    }
    //如果tial==null 说明等待队列还没有初始化. 需要自己帮忙初始化等待队列. 第一个线程获得锁的时候改完state就完成了. 不会初始化队列 
    // 第二个加锁的线程, 发现队列中没有排队的. 但是cas改变状态又失败了. 说明自己是第二个加锁, 自己需要初始化队列. 初始化一个null node当head. 自己排在他的后面,
    // 在aqs等待队列中, 对头head 代表当期锁的执行线程. 不代表排队. 第二个才算是排队的. 即tail != null的时候
    enq(node);
    return node;
}

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

//从等待列表中获得node的前节点, 果然获取不到. 就要确保前置节点能唤醒自己的情况下. 自己进入阻塞状态,
// 正常情况下: 该方法会一直阻塞当期线程. 知道获得锁才返回. 但是如果执行过程中,抛异常(tryAcquire方法) 那么会将当期节点移除等待队列. 继续向上抛出异常.
  final boolean acquireQueued(final Node node, int arg) {
        boolean failed = true;
        try {
            boolean interrupted = false;
            for (;;) {
                // 获得node的前一个节点p
                final Node p = node.predecessor();
                // 如果p是首节点, 然后去尝试获得锁. 因为TA在持有锁, 这里返回false 
                if (p == head && tryAcquire(arg)) {
                    // 将自己设置为head 表示自己占用锁
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                // 判断是否需要阻塞线程. 也就是判断TB是否要阻塞
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
      int ws = pred.waitStatus;//获得node的前节点的等待状态
      if (ws == Node.SIGNAL) // 如果是SIGNAL(-1): 后置节点需要唤醒. 自己就可以放心阻塞了. 将来会唤醒我的.
          /*
           * This node has already set status asking a release
           * to signal it, so it can safely park.
           */
          return true;
      if (ws > 0) { // 也就是CANCELLED 取消状态. 说明前置节点因为中断或者取消. 需要在等待队列中移除
          /*
           * Predecessor was cancelled. Skip over predecessors and
           * indicate retry.
           * 知道遇到前置节点的状态小于等为止.
           */
          do {
              node.prev = pred = pred.prev;
          } while (pred.waitStatus > 0);
          // 将前置节点的后置节点设置成自己.
          pred.next = node;
      } else { //如果既不是SIGNAL也不是CANCELED. 那么就将前置节点的waitStatus设置成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;
  }
  //阻塞在这里. 并检查线程是否被中断.
  private final boolean parkAndCheckInterrupt() {
      LockSupport.park(this);
      return Thread.interrupted();
  }

public void unlock() {
    sync.release(1);
}
// arg =1
public final boolean release(int arg) {
    if (tryRelease(arg)) {
        Node h = head;
        // h要值. 表示有线程在等待. h.waitStatus !=0 表示h不是新加入的节点. 在h!=null的情况. waitStatus==0 那么说明后置节点在唤醒中
        if (h != null && h.waitStatus != 0)
            unparkSuccessor(h); //释放成功, 则唤醒首节点.
        return true;
    }
    return false;
}

 protected final boolean tryRelease(int releases) {
        int c = getState() - releases; // 将当期状态-1
        //获得锁的线程和释放锁的线程不一样. 抛异常
        if (Thread.currentThread() != getExclusiveOwnerThread())
            throw new IllegalMonitorStateException();
        boolean free = false;
        // c ==0 表示锁处于空闲状态了
        if (c == 0) {
            free = true;
            // 将锁的拥有者设置为null
            setExclusiveOwnerThread(null);
        }
        // 将状态设置为 更新状态值
        setState(c);
        return free;
}

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); // 将状态设置为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; //得到后置节点
    //当后置节点为null或者s.waitStats=1 也就是取消状态时
    if (s == null || s.waitStatus > 0) {
        s = null;
        //从尾部开始向前遍历. 直到找到一个waitStatus<=0的. 并将其唤醒
        for (Node t = tail; t != null && t != node; t = t.prev)
            if (t.waitStatus <= 0)
                s = t;
    }
    //正常状态下 直接唤醒后置节点的线程
    if (s != null)
        LockSupport.unpark(s.thread);
}

    //继续从这里开始执行 .返回中断标识, 表示自己在阻塞期间没有被状态过.
 private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this);
        return Thread.interrupted();
    }

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

final Node p = node.predecessor();
    //唤醒线程的节点的前置节点是head, 尝试获取锁. 能成功的
    //tryAcquire加锁逻辑和A的加锁逻辑一样  1) 获得当期线程 2)获得状态 3 状态==0 判断是否存在排队的前辈 设置状态为1, 将锁的拥有者设置为自己
    if (p == head && tryAcquire(arg)) {
        // 将直接的节点设置为Head
        setHead(node);
        p.next = null; // help GC
        failed = false;
        return interrupted;
    }

final void lock() {
    // 1直接尝试获取锁. 不管等待队列
    if (compareAndSetState(0, 1))
        setExclusiveOwnerThread(Thread.currentThread());
    else
        acquire(1);
}

 final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        //这里和公平锁相比缺少了是否有线程排在自己之前等待(hasQueuedPredecessors)判断. 
        if (compareAndSetState(0, acquires)) {
            setExclusiveOwnerThread(current);
            return true;
        }
    }
    else if (current == getExclusiveOwnerThread()) {
        int nextc = c + acquires;
        if (nextc < 0) // overflow
            throw new Error("Maximum lock count exceeded");
        setState(nextc);
        return true;
    }
    return false;
}

public CountDownLatch(int count) {
    if (count < 0) throw new IllegalArgumentException("count < 0");
    this.sync = new Sync(count);
}

Sync(int count) {
        setState(count);
}

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

 public final void acquireSharedInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted()) //响应线程中断
        throw new InterruptedException();
    if (tryAcquireShared(arg) < 0) // 尝试获取锁. <0 就是失败 ,>0 成功, 0 无锁状态. 这里当然是失败的 以为我们初始化的状态count =1.
        doAcquireSharedInterruptibly(arg);
}


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

private void doAcquireSharedInterruptibly(int arg)
       throws InterruptedException {
       final Node node = addWaiter(Node.SHARED); //将当期线程加入等待队列尾部. 和独占锁的流程一样, 区别在于mode是共享锁类型
       boolean failed = true;
       try {
           for (;;) {
               // node也就是当期节点的前置节点
               final Node p = node.predecessor();
               if (p == head) { // 当期节点的前置节点如果是head
                   int r = tryAcquireShared(arg); // 尝试获得共享锁 这里当然也是失败的
                   if (r >= 0) {
                       //成功 键
                       setHeadAndPropagate(node, r);
                       p.next = null; // help GC
                       failed = false;
                       return;
                   }
               }
               // 如果node的前置节点不是head节点. or 获取共享锁失败.
               if (shouldParkAfterFailedAcquire(p, node) &&
                   parkAndCheckInterrupt())
                   throw new InterruptedException();
           }
       } finally {
           if (failed)
               cancelAcquire(node);
       }
   }
   //这个方法和独占锁一样, 都是判断自己是否要阻塞
    private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
       int ws = pred.waitStatus;// 前置节点的等待状态
       if (ws == Node.SIGNAL) // SIGNAL的化 自己就可以安心等待被唤醒了
           /*
            * This node has already set status asking a release
            * to signal it, so it can safely park.
            */
           return true;
       if (ws > 0) { // 如果是CANCELED. 进入循环向前查找. 知道找到一个<0的节点
           /*
            * Predecessor was cancelled. Skip over predecessors and
            * indicate retry.
            */
           do {
               node.prev = pred = pred.prev;
           } while (pred.waitStatus > 0);
           pred.next = node;
       } else {  //都不是  也就是锁状态可能是 0 -2 -3 
           /*
            * 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);  // 将前置节点设置为SIGNAL. 
       }
       return false;
   }
   //将自己阻塞掉. 唤醒时会返回中断状态. 
   private final boolean parkAndCheckInterrupt() {
       LockSupport.park(this);
       return Thread.interrupted();
   }

public void countDown() {
    sync.releaseShared(1); //释放共享锁
}

public final boolean releaseShared(int arg) {
    if (tryReleaseShared(arg)) { //尝试一次释放锁
        doReleaseShared();
        return true;
    }
    return false;
}
protected boolean tryReleaseShared(int releases) {
        // Decrement count; signal when transition to zero
        // 自旋的释放锁 知道count ==0
        for (;;) {
            int c = getState(); // 获取共享锁状态
            if (c == 0)
                return false;
            int nextc = c-1;
            if (compareAndSetState(c, nextc))
                return nextc == 0;
        }
}
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 (;;) { // 唤醒下一个等待的节点. 自旋是为了保证state能够设置成功.
        Node h = head; // 拿到首节点
        if (h != null && h != tail) { //哨兵
            int ws = h.waitStatus; // 获得首节点的等待状态
            if (ws == Node.SIGNAL) { // ==SIGNAL  CAS的将状态设置为0  直到成功
                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;
    }
}
//唤醒逻辑 
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) // 小于0 将其设置为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) { //如果是null或者 是CANCELED. 从尾部开始向前遍历. 直到找到下于等于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);
}

private void doAcquireSharedInterruptibly(int arg)
        throws InterruptedException {
        final Node node = addWaiter(Node.SHARED); //将当期线程加入等待队列尾部. 和独占锁的流程一样, 区别在于mode是共享锁类型
        boolean failed = true;
        try {
            for (;;) {
                // node也就是当期节点的前置节点
                final Node p = node.predecessor();
                if (p == head) { // 当期节点的前置节点如果是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);
        }
    }

//将当期节点设置为head. 并尝试唤醒并释放后续节点
 private void setHeadAndPropagate(Node node, int propagate) {
        Node h = head; // Record old head for check below 
        setHead(node);// 将自己设置为head
        /*
         * 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;
            //唤醒后置节点. 也就是唤醒B
            if (s == null || s.isShared())
                doReleaseShared();
        }
    }