Java concurrency – AbstractQueuedSynchronizer (AQS) synchronizer

Introduction to the

Before this introduction is already, in the ReentrantReadWriteLock sync attribute, and the sync is inherited AQS AbstractQueuedSynchronizer synchronizer. AQS adopts the template design pattern, calls its template methods (exclusive get and release synchronization state, shared get and release synchronization state and query waiting threads in the synchronization queue), overrides the specified method, we can use AQS to construct customized synchronization components.

AQS parsing

Important attributes

Private TRANSIENT volatile Node head; Private TRANSIENT volatile Node tail; private transient volatile Node tail; // Synchronization status private volatile int state;Copy the code

AQS internally defines a FIFO queue through head and tail. State indicates the synchronization state (0 indicates that no thread has acquired the synchronization state or lock, and greater than 0 indicates that there is thread possession). Both are volatile to ensure the visibility of memory

Important inner class

Static final class Node {/** SHARED mode */ static final Node SHARED = new Node(); /** EXCLUSIVE mode */ static final Node EXCLUSIVE = null; /** Because the thread waiting in the synchronization queue timed out or was interrupted, it needs to cancel the wait from the synchronization queue.The node does not change after entering this state*/ static final int CANCELLED = 1; Static final int SIGNAL = -1; static final int SIGNAL = -1; static final int SIGNAL = -1; /** * The node is in a wait queue, and the node thread is waiting on the Condtion. When another thread calls signal() on the Condtion, the node willMove from the wait queue to the synchronous queueStatic final int CONDITION = -1; static final int CONDITION = -1; PROPAGATE = PROPAGATE; PROPAGATE = PROPAGATE; PROPAGATE = PROPAGATE; /** current node waitStatus */ volatile int waitStatus; /** Precursors */ volatile Node prev; /** Next Node */ volatile Node next; /** The Thread associated with the node */ volatile Thread Thread; /** * This field is a SHARED constant if the current Node is SHARED, i.e. the Node type (exclusive and SHARED) * and the successor Node in the wait queue share the same field */ Node nextWaiter; }Copy the code

Node node is the basis of the synchronization queue. Pre and Next maintain a bidirectional queue successively. The structure of the synchronization queue is shown as follows:

When one thread succeeds in acquiring the synchronization state (or lock), other threads cannot acquire the synchronization state and are instead constructed as nodes and added to the synchronization queue

Exclusive get and release synchronization state

Exclusive access to the synchronization status

If a thread fails to obtain the synchronization status and enters the synchronization queue, the thread will not be removed from the synchronization queue when it is interrupted

public final void acquire(int arg) { if (! tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }Copy the code

TryAcquire () is called to ensure that the thread can safely acquire the synchronization state (or lock). This method is implemented by its own synchronizer. If it fails, a synchronization node is constructed and added to the end of the synchronization queue by calling addWaiter(). Finally, call acquireQueued() in an infinite loop to get the synchronization status ④. If you do not get, call shouldParkAfterFailedAcquire () method to determine whether need to block, if return true congestion nodes of threads, precursor node can rely on a team or block the thread is interrupted to awaken the blocked thread

tryAcquire

Only exceptions are thrown inside the tryAcquire() body. This method must be overridden if the custom synchronizer is to get synchronization state exclusively

    protected boolean tryAcquire(int arg) {
        throw new UnsupportedOperationException();
    }
Copy the code

addWaiter

The node is added to the synchronization queue

Private Node addWaiter(Node mode) {// Create a new Node Node Node = new Node(thread.currentThread (), mode); // CAS quickly tries to insert the tail Node. if (pred ! = null) { node.prev = pred; if (compareAndSetTail(pred, node)) { pred.next = node; return node; }} // Multiple attempts to enq(node); return node; }Copy the code

If the queue is empty or the CAS setting fails, enQ spin is set again

Private Node enq(final Node Node) {// Loop for (;;) {// get Node t = tail; If (t == null) {// Must initialize // cas set head if (compareAndSetHead(new Node())) // set tail = head; } else {// CAS set node.prev = t; if (compareAndSetTail(t, node)) { t.next = node; return t; }}}}Copy the code

It can be found from the source code that if the synchronization queue fails to add nodes, there will be an endless loop until it is successfully added

acquireQueued

After the node enters the synchronization queue, it enters a spin process. When the condition is met and the synchronization state is obtained, it can exit from the spin process, otherwise it will continue to execute

final boolean acquireQueued(final Node node, int arg) { boolean failed = true; // Interrupt token 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; } / / to determine whether a thread to block the if (shouldParkAfterFailedAcquire (p, node) && parkAndCheckInterrupt ()) interrupted = true; } } finally { if (failed) cancelAcquire(node); }}Copy the code

It can be found from the source code that only the precursor node of the current node is the head node can attempt to obtain the synchronization state, for the following reasons: ①. The head node is the node that has successfully obtained the synchronization status. After releasing the synchronization status, the head node wakes up its successor nodes. After waking up, the successor node needs to check whether it is the head node (2). Keep the FIFO synchronous queue principle

blocking

After joining the queue, the thread spins continuously to obtain synchronization status, but during the spinning process, it needs to determine whether the current thread needs to block

if (shouldParkAfterFailedAcquire(p, node) && parkAndCheckInterrupt()) interrupted = true; Private static Boolean shouldParkAfterFailedAcquire (Node Mr Pred, Node Node) {/ / precursor nodes wait states int ws = Mr Pred. WaitStatus; If (ws == Node.SIGNAL) return true; if (ws == node. SIGNAL) return true; if (ws == node. SIGNAL) return true; If (ws >0) {if (ws >0) {if (ws >0) { Do {node.prev = pred = pred.prev; } while (pred.waitStatus > 0); pred.next = node; // CAS sets the status of the precursor to SINNAL compareAndSetWaitStatus(pred, ws, node.signal); // CAS sets the status of the precursor to SINNAL compareAndSetWaitStatus(pred, ws, node.signal); } return false; }Copy the code

If shouldParkAfterFailedAcquire returns true, it will call parkAndCheckInterrupt method, the main method of internal call LockSupport tool class park blocking threads () method

private final boolean parkAndCheckInterrupt() { LockSupport.park(this); // Return thread.interrupted (); }Copy the code

Acquire () method flow

Exclusive release synchronization state

Public final Boolean release(int arg) {if (tryRelease(arg)) {Node h = head; if (h ! = null && h.waitStatus ! If (0) // Wake up the unparkprecursor (h); return true; } return false; }
Private void unparksucceeded (Node Node) {// Get the current wait state int ws = node.waitStatus; If (ws < 0) compareAndSetWaitStatus(node, ws, 0); Node s = node.next; / / if subsequent node is null or its status to CANCELLED (wait for supermarket or interruption) if (s = = null | | s. aitStatus > 0) {s = null; For (Node t = tail; t ! = null && t ! = node; t = t.prev) if (t.waitStatus <= 0) s = t; } if (s ! = null) // Wake up the thread associated with the node locksupport.unpark (s.read); }Copy the code
Copy the code


From the source, you can see that the awakened node is traversed from the end rather than the beginning, because the successor of the current node may be null, wait timeout, or interrupted, so it is traversed from the end forward
Shared synchronization state acquisition and release
The main difference between shared and exclusive access is that multiple threads can obtain a synchronization state at the same time. For example, a read lock in ReentrantReadWriteLock is used
Obtaining synchronization status in shared mode
Public final void acquireShared(int arg) {if (tryAcquireShared(arg) < 0) doAcquireShared(arg); // If (tryAcquireShared(arg) < 0) doAcquireShared(arg); }Copy the code
Private void doAcquireShared(int arg) {// Add the SHARED Node to the queue final Node Node = addWaiter(node.shared); boolean failed = true; // Interrupt token Boolean interrupted = false; // loop for (;;) Final Node p = node.predecessor(); Int r = tryAcquireShared(arg); // If (p == head) {int r = tryAcquireShared(arg); Propagate(node, r); propagate (node, r); p.next = null; // help GC if (interrupted) selfInterrupt(); failed = false; return; }} / / to determine whether a thread to block the if (shouldParkAfterFailedAcquire (p, node) && parkAndCheckInterrupt ()) interrupted = true; } } finally { if (failed) cancelAcquire(node); }}Copy the code

 
Release synchronization state in shared mode

    public final boolean releaseShared(int arg) {
        if (tryReleaseShared(arg)) {
            doReleaseShared();
            return true;
        }
        return false;
    }
Copy the code
   The CAS service is used. If the CAS operation fails, the spin cycle is performed again 
The synchronization status is obtained due to timeout
Using the built-in lock synchronized may cause deadlocks, but AQS provides timeout synchronization, that is, obtaining synchronization status within a specified period of time
Exclusive timeout gets synchronization status
public final void acquireInterruptibly(int arg) throws InterruptedException { if (Thread.interrupted()) throw new InterruptedException(); if (! tryAcquire(arg)) doAcquireInterruptibly(arg); }Copy the code
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()) throw new InterruptedException(); } } finally { if (failed) cancelAcquire(node); }}Copy the code



   Timeout control 

    public final boolean tryAcquireNanos(int arg, long nanosTimeout)
            throws InterruptedException {
        if (Thread.interrupted())
            throw new InterruptedException();
        return tryAcquire(arg) ||
            doAcquireNanos(arg, nanosTimeout);
    }
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 {// loop for (;;) Final Node p = node.predecessor(); If (p == head && tryAcquire(arg)) {setHead(node); p.next = null; // help GC failed = false; return true; } // 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); }}Copy the code
Copy the code




   

 
Shared timeout obtains synchronization status
Thank you
The Art of Concurrent Programming in Java