State

  • Internally, maintains a volatile int state
    • state = 0 is free, 1 acquired, > 1 num of reentrants
    • State is updated by unsafe.compareAndSwapInt

Node and CLH lock

In CLH lock, applying thread can only self-spin on local variable, and keep polling the previous node’s status, if it discovers that the prev node released lock, it will finish self spin

static final class Node{
        volatile int waitStatus; //control blocking
        volatile Node prev;
        volatile Node next;
        volatile Thread thread;
        Node nextWaiter; //Uses sentinel value to detect if it is shared or exclusive
}
  • if prev is cancelled, this node can use prev node’s status
  • head inside AQS is a sentinel value
  • Custom synchroinzer need to implement how state is acquired and released, q maintenance is done by the AQS already
  • WaitStatus
    • 1 - cancelled
    • -1 - signal, i.e., current thread need to wake up the next when release or cancel
    • -2 - condition, waiting to be waken up by the condition
    • -3 - propergate, shared lock

Acquire

  • You need to implement your own tryAcquire or tryAcquireShared

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

    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)) { //uses unsafe
                pred.next = node;
                return node;
            }
        }
        enq(node);
        return node;
    }

        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.
             */
            compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
        }
        return false;
    }

    private final boolean parkAndCheckInterrupt() {
        LockSupport.park(this); //to wake up thread, use unpark() or interrupted()
        return Thread.interrupted(); //ignores inturrptions during the park(), will add a self-interruption only after resource is acquired
    }

    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)) {//it is possible to get lock now, e.g., head has been released
                    setHead(node);
                    p.next = null; // help GC
                    failed = false;
                    return interrupted;
                }
                if (shouldParkAfterFailedAcquire(p, node) &&
                    parkAndCheckInterrupt())
                    interrupted = true;
            }
        } finally {
            if (failed)
                cancelAcquire(node);
        }
    }

      public void acquire(int arg) {
           if(!tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))
                      selfInterrupt(); //if the thread has been interrupted during the wait, it won't respond, here we do a self-interrupt to make up for it
      }

      public void acquireShared(int arg){
              if(tryAcquireShared(arg) < 0)
                      doAcquireShared(arg);
      }
  • Difference with fair and biased: on biased, CAS to acq first, and acq only when failed, and won’t detect if CLH is idle
  • Shared lock: when it becomes head, it will wake up the next thread, upon releasing, S lock will wake up other threads regardless of state, but X lock will wake up only when state = 0
  • for X lock, each node CAS checks if the previosu node is header, if so, try to acquire lock
  • CountDownLatch: state set to N, at each countDown(), state will CAS to dec by 1, when state= 0, unpark the main thread