Java small white series (12) : abstract queue synchronizer (AbstractQueuedSynchronizer)

One, foreword

Abstract queue synchronizer, an abstract class that uses the design pattern Java Design Pattern II: Template Pattern for template methods. At the same time, it also has a built-in CLH queue “small white 10: CLH”, to join the thread spin, when the thread (nodes) precursor node to get the lock, the thread blocks, waiting for the precursor nodes wakes up, and thread way of awakening is implemented by means of thread interrupts, about thread interruption, you can see attached: the small white 11: Thread interruption. Since AQS is an abstract class, this paper only analyzes the core process. When it comes to methods that need to be implemented by subclasses, we will skip them first, and then make an overall analysis based on specific subclasses.

Second, exclusive and sharing

When we learn CLH, we have mentioned that a node may be in exclusive mode or shared mode. However, for AQS, any thread that does not acquire the lock must be added to the CLH queue of AQS. Therefore, AQS does not particularly care about the mode of the thread that acquires the lock. We need to keep this in mind when we use the template design pattern.

Although AQS doesn’t care, it gives an optional set of methods that require subclasses to override:

• If the subclass is exclusive mode, simply override and implement exclusive mode related methods (e.g., write locks);
• If the subclass is shared mode, simply reload and implement shared mode related methods (e.g., read locks);
• What if it all happens? This type of lock is a combination of the two above (read/write lock);

From this note, we can learn several things:

• GetState: getState;
• Modify state: setState/compareAndSetState;
• The other five methods require selective overloading based on the actual situation;
• The implementation of optional methods needs to ensure that internal threads are safe, fast and not blocked;
• Other methods in AQS are not allowed to be overridden because they are final;

Therefore, AQS provides several alternative methods that need to be implemented:

private static final class LockSync extends AbstractQueuedSynchronizer { protected LockSync() { super(); } @Override protected boolean isHeldExclusively() { return super.isHeldExclusively(); } @Override protected boolean tryAcquire(int arg) { return super.tryAcquire(arg); } @Override protected boolean tryRelease(int arg) { return super.tryRelease(arg); } @Override protected int tryAcquireShared(int arg) { return super.tryAcquireShared(arg); } @Override protected boolean tryReleaseShared(int arg) { return super.tryReleaseShared(arg); }}Copy the code

Let’s analyze the implications of these methods:

• IsHeldExclusively: Indicates whether the thread is monopolizing resources. Only Condition needs to be implemented;
• TryAcquire: Exclusive mode. Arg is the number of times the lock was acquired, and returns True on success and False on failure.
• TryRelease: Exclusive mode. Arg is the number of times the lock was released, and returns True on success or False on failure.
• TryAcquireShared: Share mode. Arg attempts to acquire resources for the number of locks;
  • Negative numbers indicate failure;
  • 0 indicates success, but no available resources are available.
  • A positive number indicates success and remaining resources.
• TryReleaseShared: share mode. Arg is the number of times the lock is released. It attempts to release the resource.

Based on the above, we will analyze it later (preview plot) :

• ReentrantLock: an exclusive lock, so tryAcquire + tryRelease + isHeldExclusively
• ReentrantReadWriteLock: a shared lock is implemented.

Iii. AQS Core process

CLH queue is used internally in our AQS to manage all locks that need to be acquired but fail to be acquired. We need to join the queue and spin locally to wait for the release of the lock.

Thus, the starting point of the process is:

• The blocked thread starts with “try to acquire the lock”;
• The occupied thread starts from “try to release the lock” after it is used up.

This is also in line with the “lock” process: lock or unlock!

Let’s look at the process of requesting and releasing an exclusive lock:

Acquire: while (! tryAcquire(arg)) { // enqueue thread if it is not already queued; // possibly block current thread; Release: if (tryRelease(arg)) {unblock the first queued thread; } (The shared mode is similar to the above, but the methods called are different; And because it is shared, when a lock is released, it causes a bunch of threads to compete.)Copy the code

3.1. Discuss the fairness and unfairness of CLH again

Don’t know whether we still remember when I was in the small white ten analysis CLH, a description of the way through words and everybody said, CLH itself is fair (strictly FIFO), however, exist unfair competition, the reason is that: the new thread is the first team again request lock, or request the lock and the fall again into the queue. The JDK source code comments also say this:

The original notes also said:

Because you can try to acquire the lock before joining the queue, the new thread has a chance to compete with the first blocked element in the queue. Of course, we can also disable this unfair competition behavior in the tryAcquire or tryAcquireShared methods. Generally, most fair synchronizers return a result based on the return of the HasqueuedToraise method in the AQS.

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

The method will be analyzed in detail later, and the results are simply analyzed now. Returning true indicates that there are queued or contending threads in the queue; Returning false indicates that the queue may be empty (or that the current contending thread is itself).

Thus, tryAcquire implements fairness by relying only on the return result of Hasqueuedtoraon. Briefly, the code can be implemented as follows:

protected boolean tryAcquire(int arg) { return ! hasQueuedPredecessors(); }Copy the code

That is: When a new thread attempts to acquire the lock, it obtains the result of hasqueuedtoraise. If hasqueuedtoraise returns true, indicating that the queue has queued elements, then tryAcquire returns false, indicating that it failed to acquire the lock!

3.2. Lock acquisition process

3.2.1, AQS. Acquire

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { /** * Acquires in exclusive mode, ignoring interrupts. Implemented * by invoking at least once {@link #tryAcquire}, * returning on success. Otherwise the thread is queued, possibly * repeatedly blocking and unblocking, invoking {@link * #tryAcquire} until success. This method can be used * to implement method {@link Lock#lock}. */ public  final void acquire(int arg) { if (! tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }}Copy the code

This method is called in exclusive mode:

• TryAcquire: Try to acquire the lock first; (It’s up to the subclass to decide if it’s fair or not)
• AddWaiter: If unsuccessful, encapsulate as a node of the CLH and join the queue;
• AcquireQueued: Waits for the thread to spin in place and then blocks after spinning for some time.
• SelfInterrupt: A thread interrupts to wake up the thread (park interrupts is useless, so unpark interrupts again).

3.2.2 AQS. AddWaiter (Key: join the team)

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { /** * Creates and enqueues nodes for current thread and given mode. */ private node addWaiter(node mode) Node Node node = new Node(Thread.currentThread(), mode); // get tail Node pred = tail; // tail ! = null, pred = tail if (pred! Prev = pred (tail node) node.prev = pred; Tail = node if (compareAndSetTail(pred, node)) {// next = node pred. Next = node; return node; }} // tail = null, indicating that the queue is not initialized // need to initialize head & tail enq(node); // Return node; }}Copy the code

• Encapsulate as a Node in exclusive mode;
• Get the tail node quickly.
  • If the tail node is null, enQ is directly called to join the queue.
  • If the tail node is not null:
    • Prev points the new node to the tail node.
    • Then CAS directly sets the new node to the tail node.
    • Next of the old tail node points to the new tail node;

3.2.3, AQS. Enq

public abstract class AbstractQueuedSynchronizer
    extends AbstractOwnableSynchronizer implements java.io.Serializable {

    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;
                }
            }
        }
    }
}
Copy the code

This method is mainly used to initialize CLH queue headers and then insert new nodes.

Let’s analyze:

• The method is an infinite loop;
• Get tail tail node;
  • If the tail node is null, head = tail = null;
  • CAS set head = tail = new Node();
• If tail is not empty, the same procedure as adding a node in addWaiter is performed;
• If the node is added successfully, the precursor node of the new node is returned.

3.2.4, aqs. acquireQueued (focus: spin -> park block -> lock return)

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { /** * Acquires in exclusive uninterruptible mode for thread already in * queue. Used by condition wait methods as well as acquire. */ final boolean acquireQueued(final Node node, int arg) { boolean failed = true; try { boolean interrupted = false; // focus: spin for (;;) Final node p = node.predecessor(); // If the precursor node is the head node and the lock is acquired, the thread will exit the loop. // Even if the precursor node is the head node, the thread will only attempt to acquire the lock. In unfair mode, it may fail and continue spinning! If (p == head && tryAcquire(arg)) {setHead(node); p.next = null; // help GC failed = false; return interrupted; } // The thread cannot spin and consume CPU all the time, can it? If the following two conditions are true, it means that the thread is in the blocking state. After the precursor node // releases the lock, By the interrupt to awaken this thread if (shouldParkAfterFailedAcquire (p, Node) && / / judge whether the state of the precursor node to node. If a SIGNAL / / shouldParkAfterFailedAcquire = true will go to the park thread / / this method is the method into the blocking state, once park, ParkAndCheckInterrupt ()) interrupted = true; // A suspended thread cannot respond to a thread interrupt, so check if it has been interrupted after unpark. If (failed) cancelAcquire(node) cancelAcquire(node); if (failed) cancelAcquire(node); }}}Copy the code

The parkAndCheckInterrupt thread is suspended and no further code is executed! Unpark until the lock is released by the precursor node! Also, thread interrupts cannot respond during suspension!

3.2.5, AQS. ShouldParkAfterFailedAcquire

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { /** * Checks and updates status for a node that failed to acquire. * Returns true if thread should block. This is the main signal * control in all acquire loops. Requires that pred == node.prev. */ private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) { int ws = pred.waitStatus; If (ws == node.signal) /* * The precursor Node has been set to this state: that is, the Node thread will notify the next Node when releasing the lock */ return true; If (ws > 0) {/* * The thread cancels */ do {// keep looking forward until the node state <= 0 // this line splits into two lines: // pred = pred.prev; // Get the precursor node; // node.prev = pred; // The prev of the new node points to the new precursor node node.prev = pred = pred.prev; } while (pred.waitStatus > 0); // Next = new node 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; }}Copy the code

The key is compareAndSetWaitStatus. Once the state of the precursor is set to SIGNAL, the node’s next spin is blocked because the waitStatus of the precursor is SIGNAL the next time the loop enters the method.

3.2.6, AQS. ParkAndCheckInterrupt (key: hang)

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { Private final Boolean parkAndCheckInterrupt() {// Call park to enter waiting state locksupport.park (this); // ----------------- suspend! // Returns whether the thread has been interrupted, and clears the interrupted state! return Thread.interrupted(); }}Copy the code

This method is simple: simply put the thread into the blocking state, and then return whether it has ever been interrupted.

At this point, the whole process of obtaining the lock is analyzed, let’s summarize:

1. TryAcquire attempts to acquire locks;
2. AddWaiter joins the team if the capture fails;
3. AcquireQueued Spins, checks or sets the precursor node waitStatus = SIGNAL;
4. Park then suspends and does not respond to interrupts.
5. Wait for the precursor node to be released to unpark (as analyzed below);
6. The next loop after unpark starts the lock race (because after unpark, it is preceded by the head node);
7. If that fails, suspend again and repeat steps 5 and 6.
8. If successful, it sets itself as head and returns whether it was ever interrupted.

It can be seen that acquire is the entrance to acquire the lock. Once it fails, the method will also be blocked, and the caller (i.e. thread) will get the lock directly, or the execution will continue for other reasons such as timeout or cancellation. The whole process is actually very simple, just some details, you need to pay attention to, where I highlight, you should carefully read the notes, savor the possibilities.

3.3 Lock release process

We have analyzed the process of acquiring locks, and also mentioned a little unpark. Therefore, our lock release process is very easy to understand and very short. The lock access is “acquire” and the lock release is “release”; For shared mode, the lock release mode is “releaseShared”.

3.3.1, AQS. Release

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { public final boolean release(int arg) { if (tryRelease(arg)) { Node h = head; if (h ! = null && h.waitStatus ! = 0) unparkSuccessor(h); return true; } return false; }}Copy the code

This method is very simple. Call the subclass “tryRelease”, which is implemented by the subclass, so we’ll skip ahead and just look at the success case. When tryRelease successfully releases the lock, the current head node is retrieved, if head.waitStatus! = 0, go to unpark; So who is unpark? The next node of unpark, of course.

3.3.2 rainfall distribution on 10-12, unparkSuccessor

public abstract class AbstractQueuedSynchronizer extends AbstractOwnableSynchronizer implements java.io.Serializable { Private void unparksucceeded (Node Node) {// Try to set your own waitStatus to 0 int ws = node.waitStatus; if (ws < 0) compareAndSetWaitStatus(node, ws, 0); // Get the next node and try to wake it up! Node s = node.next; / / if the next node does not exist, or have been cancelled the if (s = = null | | s. aitStatus > 0) {s = null; For (node t = tail; for (node t = tail; t ! = null && t ! = node; t = t.prev) if (t.waitStatus <= 0) s = t; } if (s ! = null) // unpark! Unsuspend thread locksupport. unpark(s.read); }}Copy the code

The whole unlock process is analyzed. Yes, it’s that simple!

Four, summary

AQS at first glance, there are many methods, but we only focus on the main process, in fact, just a few functions, very simple; Hopefully, after you have learned and mastered the main process, you will understand the other methods and there will be no pressure. AQS analysis finished, next, I will lead you to analyze some of the commonly used subclasses, such as: ReentrantLock, ReadWriteLock and so on.