Java并发-AQS(1)

class Mutex implements Lock {
  // 静态内部类，自定义同步器
  private static class Sync extends AbstractQueuedSynchronizer {
		
    // 当状态为0时 获取锁
    @Override
    protected boolean tryAcquire(int arg) {
      if (compareAndSetState(0, arg)) {
        setExclusiveOwnerThread(Thread.currentThread());
        return true;
      }
      return false;
    }

    // 释放锁 将状态设置为0
    @Override
    protected boolean tryRelease(int arg) {
      if (getState() == 0) {
        throw new IllegalMonitorStateException();
      }
      setExclusiveOwnerThread(null);
      setState(0);
      return true;
    }

    // 是否处于占用状态
    @Override
    protected boolean isHeldExclusively() {
      return getState() == 1;
    }
    
    // 返回一个Condition 每个condition都包含一个condition队列
    Condition newCondition() {
      return new ConditionObject();
    }
  }
  
  // 仅需要将所有的操作代理到Sync上即可
  private final Sync sync = new Sync();
  
  @Override
  public void lock() {
		sync.acquire(1);
  }
  
  @Override
  public boolean tryLock() {
    return sync.tryAcquire(1);
  }
  
  @Override
  public void unlock() {
    sync.release(1);
  }
  
  @Override
  public Condition newCondition() {
    return sync.newCondition();
  }
  
  @Override
  public void lockInterruptibly() throws InterruptedException {
    sync.acquireInterruptibly(1);
  }
  
  @Override
  public boolean tryLock(long time, TimeUnit unit) throws InterruptedException {
    return sync.tryAcquireNanos(1, unit.toNanos(time));
  }
}

public final void acquire(int arg) {
  if (!tryAcquire(arg) &&
      // acquireQueued 返回线程是否被中断
      acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
    // 中断当前线程
    selfInterrupt();
}

static void selfInterrupt() {
  Thread.currentThread().interrupt();
}

private Node addWaiter(Node mode) {
  // 将当前线程包装成Node
  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)) {
      // 将前驱节点的next设置为自己
      pred.next = node;
      return node;
    }
  }
  //
  enq(node);
  return node;
}

private Node enq(final Node node) {
  for (;;) {
    Node t = tail;
    // 队列为空，延迟初始化
    if (t == null) { // Must initialize
      // 初始化时 新增一个dummy节点
      if (compareAndSetHead(new Node()))
        tail = head;
    } else {
      node.prev = t;
      if (compareAndSetTail(t, node)) {
        t.next = node;
        return t;
      }
    }
  }
}

final boolean acquireQueued(final Node node, int arg) {
  boolean failed = true;
  try {
    boolean interrupted = false;
    for (;;) {
      final Node p = node.predecessor();
      // 当前节点的前驱节点为head时，再次尝试获取锁
      if (p == head && tryAcquire(arg)) {
        // head向后挪一位，并将head设置为dummy节点
        setHead(node);
        p.next = null; // help GC
        failed = false;
        return interrupted;
      }
      // 获取锁失败时，判断是否需要将当前线程挂起 && 挂起并检查线程中断状态
      if (shouldParkAfterFailedAcquire(p, node) &&
          parkAndCheckInterrupt())
        // 线程被中断
        interrupted = true;
    }
  } finally {
    if (failed)
      cancelAcquire(node);
  }
}

private void setHead(Node node) {
  head = node;
  node.thread = null;
  node.prev = null;
}

private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
  int ws = pred.waitStatus;
  if (ws == Node.SIGNAL)
    /*
     * This node has already set status asking a release
     * to signal it, so it can safely park.
     */
    // 这个节点已经设置状态，释放时发出信号，可以被安全的挂起
    return true;
  if (ws > 0) {
    /*
     * Predecessor was cancelled. Skip over predecessors and
     * indicate retry.
     */
    // 前驱节点已被取消(超时/中断)。跳过这些节点，直接排在等待锁的节点后面
    do {
      node.prev = pred = pred.prev;
    } while (pred.waitStatus > 0);
    pred.next = node;
  } else {
    /*
     * waitStatus must be 0 or PROPAGATE.  Indicate that we
     * need a signal, but don't park yet.  Caller will need to
     * retry to make sure it cannot acquire before parking.
     */
    // 前驱既不是CANCELED也不是SIGNAL。使用CAS将前驱节点设置为SIGNAL状态。
    compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
  }
  return false;
}

private final boolean parkAndCheckInterrupt() {
  // 将线程挂起
  LockSupport.park(this);
  // 被唤醒时，检查线程的中断状态
  return Thread.interrupted();
}

public final boolean release(int arg) {
  if (tryRelease(arg)) {
    Node h = head;
    // 是否有后继节点需要唤醒。
    // head为懒初始化，没有排队的线程则为null
    // 节点状态默认为0，当有需要被唤醒的节点时，会将其前驱节点状态设置为SIGNAL(-1)
    if (h != null && h.waitStatus != 0)
      unparkSuccessor(h);
    return true;
  }
  return false;
}

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;
  // head节点状态<0，则置为0
  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;
  // 后继节点为空，或者节点状态>0(超时/中断取消)时，按FIFO找到最近需要唤醒的节点
  if (s == null || s.waitStatus > 0) {
    s = null;
    // 从队列尾部向前遍历至head，找到距离head最近且节点状态<=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);
}

public final void acquireShared(int arg) {
  if (tryAcquireShared(arg) < 0)
    doAcquireShared(arg);
}

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);
        // 返回值小于0，获取共享锁失败
        // 返回值等于0，获取共享锁成功，后续获取会失败
        // 返回值大于0，获取共享锁成功，后续获取很可能成功
        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);
  }
}

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

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

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 (;;) {
        AbstractQueuedSynchronizer.Node h = head;
        if (h != null && h != tail) {
            int ws = h.waitStatus;
          	// 头节点ws==SIGNAL，说明后继节点需要被唤醒
            if (ws == AbstractQueuedSynchronizer.Node.SIGNAL) {
              	// 此处CAS保证下面的unparkSuccessor(h)只会被执行一次
                if (!compareAndSetWaitStatus(h, AbstractQueuedSynchronizer.Node.SIGNAL, 0))
                    continue;            // loop to recheck cases
                unparkSuccessor(h);
            }
          	// ws为0的情况：①上面的if CAS操作；②节点新增时默认值(尾节点)；
          	// CAS操作失败，说明ws被修改，即有新节点加入。再循环一次，即可直接唤醒。
            else if (ws == 0 &&
                    !compareAndSetWaitStatus(h, 0, AbstractQueuedSynchronizer.Node.PROPAGATE))
                continue;                // loop on failed CAS
        }
        if (h == head)                   // loop if head changed
            break;
    }
}

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;
      }
      // 超时时间减去当前时间，小于等于0则表示到期/超时
      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);
  }
}