Thoroughly understand AbstractQueuedSynchronizer (2)

ReentrantLock defaults to unfair locking unless you pass true in the constructor.

public ReentrantLock(a) {
    // Default unfair lock
    sync = new NonfairSync();
}
public ReentrantLock(boolean fair) {
    sync = fair ? new FairSync() : new NonfairSync();
}
Copy the code

Lock method for fair locking:

static final class FairSync extends Sync {
    final void lock(a) {
        acquire(1);
    }
    // AbstractQueuedSynchronizer.acquire(int arg)
    public final void acquire(int arg) {
        if(! tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }protected final boolean tryAcquire(int acquires) {
        final Thread current = Thread.currentThread();
        int c = getState();
        if (c == 0) {
            // 1. In contrast to unfair locks, there is one more judgment: whether there is a thread waiting
            if(! hasQueuedPredecessors() && compareAndSetState(0, acquires)) {
                setExclusiveOwnerThread(current);
                return true; }}else if (current == getExclusiveOwnerThread()) {
            int nextc = c + acquires;
            if (nextc < 0)
                throw new Error("Maximum lock count exceeded");
            setState(nextc);
            return true;
        }
        return false; }}Copy the code

The lock method for an unfair lock:

static final class NonfairSync extends Sync {
    final void lock(a) {
        // 2. In contrast to a fair lock, CAS is performed directly first, and success is returned
        if (compareAndSetState(0.1))
            setExclusiveOwnerThread(Thread.currentThread());
        else
            acquire(1);
    }
    // AbstractQueuedSynchronizer.acquire(int arg)
    public final void acquire(int arg) {
        if(! tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }protected final boolean tryAcquire(int acquires) {
        returnnonfairTryAcquire(acquires); }}/** * Performs non-fair tryLock. tryAcquire is implemented in * subclasses, but both need nonfair try for trylock method. */
final boolean nonfairTryAcquire(int acquires) {
    final Thread current = Thread.currentThread();
    int c = getState();
    if (c == 0) {
        // There is no judgment on blocking queues
        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;
}
Copy the code

Summary: There are only two differences between fair and unfair locks:

• When an unfair lock is called, CAS will be called to grab the lock. If the lock is not occupied at this time, CAS will return the lock.
• After a CAS failure, an unfair lock will enter the tryAcquire method just like a fair lock. In the tryAcquire method, if the lock is released (state == 0), the unjust lock will be directly seized by CAS. However, fair lock will determine whether there is a thread waiting in the queue, if there is, it will not grab the lock, obediently queue to the back.

There are only two differences between a fair lock and an unfair lock. If both CAS are unsuccessful, then the unfair lock is the same as the fair lock.

Relatively speaking, unfair locks have better performance because of their higher throughput. Of course, unfair locking makes the timing of lock acquisition more uncertain and can lead to chronically hungry threads in the blocking queue.

AddConditionWaiter () adds the current node to the conditional queue, which is thread-safe.

// Enqueue the node corresponding to the current thread and insert it at the end of the queue
private Node addConditionWaiter(a) {
    Node t = lastWaiter;
    // If the last node of the conditional queue is cancelled, it is cleared
    // If waitStatus is not equal to node. CONDITION, this Node is unqueued.
    if(t ! =null&& t.waitStatus ! = Node.CONDITION) {// This method iterates through the conditional queue and clears all cancelled nodes from the queue
        unlinkCancelledWaiters();
        t = lastWaiter;
    }
    // When node is initialized, specify waitStatus as Node.condition
    Node node = new Node(Thread.currentThread(), Node.CONDITION);

    // t is now lastWaiter, the end of the queue
    // If queue is empty
    if (t == null)
        firstWaiter = node;
    else
        t.nextWaiter = node;
    lastWaiter = node;
    return node;
}
Copy the code

The above piece of code is simple enough to put the current thread at the end of the conditional queue. In the addWaiter method, there is an unlinkCancelledWaiters() method, which is used to clear nodes from the queue that have been unwaited for. This method is called once if a cancel operation occurs while await (this will be said later), or if the last node is found to be cancelled while the node is being queued.

// The wait queue is a one-way linked list, traversing the list to clear out the nodes whose wait has been cancelled
// This is a list operation
private void unlinkCancelledWaiters(a) {
    Node t = firstWaiter;
    Node trail = null;
    while(t ! =null) {
        Node next = t.nextWaiter;
        // If the Node is not in node. CONDITION, the Node is cancelled
        if(t.waitStatus ! = Node.CONDITION) { t.nextWaiter =null;
            if (trail == null)
                firstWaiter = next;
            else
                trail.nextWaiter = next;
            if (next == null)
                lastWaiter = trail;
        }
        elsetrail = t; t = next; }}Copy the code

So once you release the lock, then you have this, and then you spin, and if you find that you’re not in the blocking queue, then you hang up and wait to be transferred to the blocking queue.

int interruptMode = 0;
// If it is not a blocking queue, it is a blocking queue
while(! isOnSyncQueue(node)) {// The thread is suspended
    LockSupport.park(this);

    // You can wait to see when it is unpark
    if((interruptMode = checkInterruptWhileWaiting(node)) ! =0)
        break;
}
Copy the code

IsOnSyncQueue (Node Node) is used to determine whether the Node has been moved to the blocking queue:

// The Node is initialized with waitStatus = node. CONDITION
// I mentioned earlier that we need to move the node from the conditional queue to the blocking queue.
// This method is to determine whether the node has moved to the blocking queue
final boolean isOnSyncQueue(Node node) {

    // The node's waitStatus is set to 0 when moved over, which will be discussed later when we say signal
    // If waitStatus is still Node.CONDITION (-2), it must still be in the CONDITION queue
    // If the node prev is still null, it is definitely not in a blocking queue.
    if (node.waitStatus == Node.CONDITION || node.prev == null)
        return false;
    // If node already has a successor node next, it must be blocking the queue
    if(node.next ! =null) 
        return true;

    // The following method starts at the end of the blocking queue and works backwards. If it finds an equal number, it is in the blocking queue, otherwise it is not

    // Node.prev ()! = null to infer that node is blocking on the queue? The answer is: no.
    // Set node.prev to tail.
    // Then the CAS operation sets itself as the new tail, but this time the CAS may fail.

    return findNodeFromTail(node);
}

// Traverse through the end of the blocking queue, returning true if found
private boolean findNodeFromTail(Node node) {
    Node t = tail;
    for (;;) {
        if (t == node)
            return true;
        if (t == null)
            return false; t = t.prev; }}Copy the code

Returning to the previous loop, isOnSyncQueue(node) returns false, then locksupport.park (this); Here the thread hangs.

After signal in the previous step, our thread is moved from the conditional queue to the blocking queue, and is ready to acquire the lock. Once the lock is retrieved, continue to execute. Wait for the thread to recover from suspension and continue reading

int interruptMode = 0;
while(! isOnSyncQueue(node)) {// The thread is suspended
    LockSupport.park(this);

    if((interruptMode = checkInterruptWhileWaiting(node)) ! =0)
        break;
}
Copy the code

InterruptMode is explained first. InterruptMode can be REINTERRUPT (1), THROW_IE (-1), or 0

• REINTERRUPT: When an await is returned, the interrupt status needs to be reset
• THROW_IE: To return an await, InterruptedException is thrown
• 0: indicates that no interrupt has occurred during await

There are three situations that will make locksupport.park (this); This return continues:

• Regular path. Signal -> Transfer node to blocking queue -> Lock acquired (unpark)
• The thread is interrupted. While in park, another thread interrupts this thread
• Signal when we said that the precursor node after the transfer was cancelled or the CAS operation to the precursor node failed
• Wake up. Object.wait() also has this problem

Thread wakes up after the first step is to call checkInterruptWhileWaiting (node) this method, this method is used to determine whether interruption happened during thread hanging, if interrupt comes up is interrupted before signal invocation, or signal after the interrupt.

// 1. If the signal has been interrupted before, return THROW_IE
// 2. If signal is followed by interrupt, return REINTERRUPT
// 3. No interrupt occurred, return 0
private int checkInterruptWhileWaiting(Node node) {
    return Thread.interrupted() ?
        (transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) :
        0;
}
Copy the code

Thread.interrupted() : If the current Thread is interrupted, this method returns true and resets the interrupted status to false, hence the use of reinterrupts.

See how to determine whether the interrupt occurred before or after signal:

// This method is called only if the thread is interrupted
// Move the unwaited node to the blocking queue if necessary
// Returns true: If this thread is canceled before signal,
final boolean transferAfterCancelledWait(Node node) {
    // Use CAS to set the node status to 0
    // If this CAS step succeeds, the interrupt occurred before signal, because if signal had occurred first, waitStatus would have been set to 0
    if (compareAndSetWaitStatus(node, Node.CONDITION, 0)) {
        // Put the node into the blocking queue
        // We can see that even if an interrupt occurs, it is still transferred to the blocking queue
        enq(node);
        return true;
    }

    // The CAS failed, and the signal method must have set waitStatus to 0
    // The signal method moves the node to the blocking queue, but it may not be done yet, and spins to wait for it to complete
    // Of course, this is rare: after a signal call, an interrupt occurs before the transfer is complete
    while(! isOnSyncQueue(node)) Thread.yield();return false;
}
Copy the code

Again, even if an interrupt occurs, the node is still moved to the blocking queue. At this point, you probably know how the while loop exits. Either an interruption or a successful transfer. This describes a scenario where there is a thread that is behind the conditional queue, but because it is interrupted, it will wake up, and then it realizes that it is not the one that was signalled, but it will voluntarily enter the blocking queue.

At this point, we can finally say what interruptMode is used for.

• 0: do nothing, not interrupted;
• The THROW_IE: await method throws InterruptedException because it represents an interruption during await();
• REINTERRUPT: Re-interrupts the current thread, since it represents an interrupt that has occurred after signal() and has not been interrupted during await()

private void reportInterruptAfterWait(int interruptMode)
    throws InterruptedException {
    if (interruptMode == THROW_IE)
        throw new InterruptedException();
    else if (interruptMode == REINTERRUPT)
        selfInterrupt();
}
Copy the code

This part of the interrupt state, you should understand it, if not, read it several times.

The last section of Condition.

public final void awaitUninterruptibly(a) {
    Node node = addConditionWaiter();
    int savedState = fullyRelease(node);
    boolean interrupted = false;
    while(! isOnSyncQueue(node)) { LockSupport.park(this);
        if (Thread.interrupted())
            interrupted = true;
    }
    if (acquireQueued(node, savedState) || interrupted)
        selfInterrupt();
}
Copy the code

Very simple, paste the code everyone will understand, I will not nonsense.

Let me draw a dividing line. Did you feel cheated? I said a lot, but it had nothing to do with cancellation. Let’s look at another lock method for ReentrantLock:

public void lockInterruptibly(a) throws InterruptedException {
    sync.acquireInterruptibly(1);
}
Copy the code

Throws InterruptedException (InterruptedException).

public final void acquireInterruptibly(int arg)
        throws InterruptedException {
    if (Thread.interrupted())
        throw new InterruptedException();
    if(! tryAcquire(arg)) doAcquireInterruptibly(arg); }Copy the code

Keep going:

private void doAcquireInterruptibly(int arg) throws InterruptedException {
    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;
            }
            if (shouldParkAfterFailedAcquire(p, node) &&
                parkAndCheckInterrupt())
                // If there is an exception, end the method immediately and throw an exception.
                // Instead of just marking the return value of this method as representing the interrupted state
                // The exception is thrown directly, and the outer layer is not caught, all the way out to lockInterruptibly
                throw newInterruptedException(); }}finally {
        // Failed is true if you use InterruptedException
        if(failed) cancelAcquire(node); }}Copy the code

While we’re at it, let’s talk about cancelAcquire:

private void cancelAcquire(Node node) {
    // Ignore if node doesn't exist
    if (node == null)
        return;
    node.thread = null;
    // Skip cancelled predecessors
    // Find a suitable precursor. It basically says "please remove" all nodes in the queue before it that have been canceled.
    Node pred = node.prev;
    while (pred.waitStatus > 0)
        node.prev = pred = pred.prev;
    // predNext is the apparent node to unsplice. CASes below will
    // fail if not, in which case, we lost race vs another cancel
    // or signal, so no further action is necessary.
    Node predNext = pred.next;
    // Can use unconditional write instead of CAS here.
    // After this atomic step, other Nodes can skip past us.
    // Before, we are free of interference from other threads.
    node.waitStatus = Node.CANCELLED;
    // If we are the tail, remove ourselves.
    if (node == tail && compareAndSetTail(node, pred)) {
        compareAndSetNext(pred, predNext, null);
    } else {
        // If successor needs signal, try to set pred's next-link
        // so it will get one. Otherwise wake it up to propagate.
        int ws;
        if(pred ! = head && ((ws = pred.waitStatus) == Node.SIGNAL || (ws <=0&& compareAndSetWaitStatus(pred, ws, Node.SIGNAL))) && pred.thread ! =null) {
            Node next = node.next;
            if(next ! =null && next.waitStatus <= 0)
                compareAndSetNext(pred, predNext, next);
        } else {
            unparkSuccessor(node);
        }
        node.next = node; // help GC}}Copy the code

There is nothing to be said for this method, which should be CANCELLED until Node CANCELLED, as long as waitStatus is set to Node.CANCELLED, lots of cases are CANCELLED out of the blocking queue, either actively or passively.

Interrupt is not a command like kill -9 PID. It does not mean that we interrupt a thread and the thread stops running. Interrupts represent thread state. Each thread has an interrupt state associated with it. This is a Boolean value of true or false, starting with false. Interrupts in Java are not the same as operating system interrupts. For interrupt states, we need to focus on the following methods in the Thread class:

// Thread instance method, holding Thread instance reference to detect Thread interrupt status
public boolean isInterrupted(a) {}

// A static method in Thread that checks whether the calling Thread has been interrupted
// Note: When this method returns the interrupted status, it resets the interrupted status of this thread to false
// So, if we call this method twice in a row, the return value of the second call must be false
public static boolean interrupted(a) {}

// An instance method of the Thread class that sets the interrupt status of a Thread to true
public void interrupt(a) {}
Copy the code

If you interrupt a thread by setting its status to true, what the interrupted thread does with this status is up to the thread itself. Such as the following code:

while(! Thread.interrupted()) { doWork(); System.out.println("I'm done with one thing, ready to move on to the next, if no other thread interrupts me.");
}
Copy the code

This code will respond to interrupts, and it will determine the interrupt state as it works. However, other interrupts are relatively rare outside of the JDK source code, which is very particular.

Of course, interrupts are more than just thread states, otherwise there is no need to develop a concept. A thread is automatically aware when it is interrupted if it is in one of three conditions:

• Join (), join(long), join(long, int), sleep(long), sleep(long), wait(long, int) Int). Throws InterruptedException. If a thread blocks on one of these methods (which, as we know, cause the current thread to block), if another thread interrupts the thread, it will immediately return from the methods, throwing InterruptedException and resetting the interrupt status to false.
• I/O blocking operations in classes that implement the InterruptibleChannel interface, such as the Connect and Receive methods in DatagramChannel. If the thread block here, interrupt threads can lead to throw these methods ClosedByInterruptException and reset the interrupt status.
• Select method in Selector, for reference to the NIO article I wrote. Once interrupted, the method returns immediately

The above three cases are the most special because they are automatically aware of the interruption (automatic, also based on the underlying implementation) and reset the interrupt state to false after doing the corresponding operation. Are there only three ways to automatically detect interruptions? No, if the thread is blocked in locksupport. park(Object obj) method, also called suspended, this interrupt will also cause the thread to wake up, but will not reset the interrupted state, so wake up to detect the interrupted state will be true.

Once an interrupt occurs and we receive this information, what do we do about the interrupt? This section will briefly examine this question. We often write code like this:

try {
    Thread.sleep(10000);
} catch (InterruptedException e) {
    // ignore
}
// go on 
Copy the code

When the sleep ends and we continue, we often don’t know if the piece of code actually slept for 10 seconds, or just slept for one second before being interrupted. The problem with this code is that we swallow the exception message. (For the sleep method, I believe that most of the time we don’t care if we interrupt, here’s an example). There are two methods of ReentrantLock:

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

public void lockInterruptibly(a) throws InterruptedException {
    sync.acquireInterruptibly(1);
}
Copy the code

As we mentioned earlier, the lock() method does not respond to interrupts. If thread1 calls the lock() method and does not grab the lock for a long time, thread2 interrupts it. Thread1 does not respond to the request and continues to grab the lock without throwing away the “interrupted” message. We can look at the following code:

public final void acquire(int arg) {
    if(! tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))// We can see that there is no special processing here, which is to record the interrupt state.
        // This way, if the outer method needs to be checked, at least we don't lose this information
        selfInterrupt();// Thread.currentThread().interrupt();
}
Copy the code

For the lockInterruptibly() method, which has a throw InterruptedException, the signal tells us that if we want to cancel the thread lock grab, we simply interrupt the thread and it will return immediately. InterruptedException is thrown.

In parallel packets, there are many examples of this kind of handling of interrupts, providing two methods: responsive interrupt and non-responsive interrupt. In the case of non-responsive interrupt, logging the interrupt rather than losing this information. For example, the two methods in Condition look like this:

void await(a) throws InterruptedException;
void awaitUninterruptibly(a);
Copy the code

In general, if the method throws InterruptedException, the first sentence in the method body is:

public final void await(a) throws InterruptedException {
    if (Thread.interrupted())
        throw newInterruptedException(); . }Copy the code

The skillful use of interrupts helps us to write elegant code and to analyze other people’s source code.