In this paper, the source AbstractQueuedSynchronizer resolution principle Reproduced please specify
Most AbstractQueuedSynchronizer is referred to as “AQS Java Lock, Semaphore, such as CountDownLatch public rely on the framework, realization depends on the first-in, first-out (FIFO) waiting queue blocking locks. To understand its code principle helps us to understand the Java Lock derived class principle, to help us develop custom Lock.
The main principle of
Node internal class properties
static final class Node {
/** marks the current node share lock mark */
static final Node SHARED = new Node();
/** marks the current node's exclusive lock mark */
static final Node EXCLUSIVE = null;
/** Interrupts or times out of lock contention */
static final int CANCELLED = 1;
/** The lock is about to be released, so you need to wake up the next fetch thread and point the head node to the next node */
static final int SIGNAL = -1;
Condition */
static final int CONDITION = -2;
/ / Shared lock
static final int PROPAGATE = -3;
// Thread wait states correspond to the above four states respectively
volatile int waitStatus;
// The front node reference
volatile Node prev;
// place a post-node reference
volatile Node next;
volatile Thread thread;
// The next node needs to wake up
Node nextWaiter;
final boolean isShared(a) {
return nextWaiter == SHARED;
}
Node() {}
/** Constructor used by addWaiter. */
Node(Node nextWaiter) {
this.nextWaiter = nextWaiter;
THREAD.set(this, Thread.currentThread());
}
/** Constructor used by addConditionWaiter. */
Node(int waitStatus) {
WAITSTATUS.set(this, waitStatus);
THREAD.set(this, Thread.currentThread());
}
Copy the codeAbstractQueuedSynchronizer internal attributes
/** * queue header */
private transient volatile Node head;
/** * queue end */
private transient volatile Node tail;
/** * synchronization status, based on this value to obtain the lock, the default thread is 0 */
private volatile int state;
protected final int getState(a) {
return state;
}
protected final void setState(int newState) {
state = newState;
}
protected final boolean compareAndSetState(int expect, int update) {
return STATE.compareAndSet(this, expect, update);
}
Copy the codeWhen you enter the ReentrantLock and acquire the lock, call the AQS method
public void lock(a) {
sync.acquire(1);
}
Copy the codeSync is already inner classes, inherit from AbstractQueuedSynchronizer, used to achieve fair and not fair lock lock.
acquire
public final void acquire(int arg) {
// Try to get the lock
if(! tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg))// Failed to obtain the lock
selfInterrupt(); // Set the current thread to interrupt
}
Copy the codeThe primary stream first acquires the lock, and if it fails, enqueue.
acquireQueued
final boolean acquireQueued(final Node node, int arg) {
boolean interrupted = false;
try {
for (;;) { / / spin
final Node p = node.predecessor(); // The front node
if (p == head && tryAcquire(arg)) {// The front node is head, and the next one about to acquire the lock tries to acquire the lock
setHead(node); // If it succeeds, point head to node
p.next = null; // help GC
return interrupted;
}
if (shouldParkAfterFailedAcquire(p, node)) // Determine whether the front-node waitStatus is SIGNAL, if not, the thread will not be suspended
interrupted |= parkAndCheckInterrupt(); // Suspend the thread. When the thread is awakened, it returns to the thread interrupt status and cleans up the interrupt}}catch (Throwable t) {
cancelAcquire(node); // Wake up the thread and remove the blocked queue
if (interrupted)
selfInterrupt();
throwt; }}Copy the codeThe acquireQueued method either successfully acquires the lock and exits the loop, or an exception occurs and enters the exception handling logic.
shouldParkAfterFailedAcquire
private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
int ws = pred.waitStatus;
if (ws == Node.SIGNAL) // The thread is ready to wake up
/* * This node has already set status asking a release * to signal it, so it can safely park. */
return true;
if (ws > 0) { // CANCELLED threads greater than 0 should cancel the lock race
do {
node.prev = pred = pred.prev;
} while (pred.waitStatus > 0);
pred.next = node; // Delete the ws = CANCELLED queue node
} else { // PROPAGATE is equal to 0 or PROPAGATE
/* * 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. */
pred.compareAndSetWaitStatus(ws, Node.SIGNAL);
}
return false;
}
Copy the codeThis method does two things: compares waitStatus or the setting, and moves the node forward. When p=head is equal, it may be head or empty, and the queue has not been initialized successfully, then the lock is acquired once. Another example is node head, or node re-entrant attempts to acquire the lock, which succeeds, and resets the head.
cancelAcquire
private void cancelAcquire(Node node) {
// Ignore if node doesn't exist
if (node == null)
return;
node.thread = null;
Node pred = node.prev;
while (pred.waitStatus > 0) // Skip the thread whose state is greater than 0
node.prev = pred = pred.prev;
Node predNext = pred.next;
// Set current node to CANCELLED to cancel the lock race
node.waitStatus = Node.CANCELLED;
// If node is tail, set the front node to tail.
if (node == tail && compareAndSetTail(node, pred)) {
pred.compareAndSetNext(predNext, null); // add tail.next = null
} else {
int ws;
// Check whether the node state is SIGNAL or head
// If you want to delete a node, you can delete it.
if(pred ! = head && ((ws = pred.waitStatus) == Node.SIGNAL || (ws <=0&& pred.compareAndSetWaitStatus(ws, Node.SIGNAL))) && pred.thread ! =null) {
Node next = node.next;
if(next ! =null && next.waitStatus <= 0)
pred.compareAndSetNext(predNext, next);// back and forth
} else {
unparkSuccessor(node); // Wake up the post-node
}
node.next = node; // help GC}}Copy the codeThe main task of this method is to find the pre-node with valid state, set the thread status CANCELLED, cancel the thread lock competition, find the post-thread, associate the before and after guidance, and delete itself from the queue. However, it is possible that the front node is head, so it is necessary to wake up the rear node, delete the CANCELLED node, and obtain the lock
unparkSuccessor
How do I wake up the thread
private void unparkSuccessor(Node node) {
int ws = node.waitStatus;
if (ws < 0) // To wake up the next thread, the SIGNAL state should be cleared
node.compareAndSetWaitStatus(ws, 0);
Node s = node.next;
if (s == null || s.waitStatus > 0) {
s = null;
for(Node p = tail; p ! = node && p ! =null; p = p.prev) // Work backwards to find the nearest node
if (p.waitStatus <= 0)
s = p;
}
if(s ! =null)
LockSupport.unpark(s.thread); // Wake up the thread
}
Copy the codeaddWaiter
private Node addWaiter(Node mode) {
Node node = new Node(mode);
for (;;) {
Node oldTail = tail;
if(oldTail ! =null) {//tail already exists. Add thread to tail
node.setPrevRelaxed(oldTail);
if (compareAndSetTail(oldTail, node)) {
oldTail.next = node;
returnnode; }}else {
initializeSyncQueue(); // Initialize the queue head, tail attributes}}}Copy the codeTo create queued nodes according to the given mode, the mode is divided into two modes, node.exclusive and Node.shared.
tryAcquire
acquire
acquire
Analysis tryAcquire
TryAcquire is the only implementation of lock acquisition, and there are mainly subclasses. Because AQS supports exclusive locks and shared locks, whether it supports reentrant or not, it is more suitable for subclasses to implement lock acquisition logic. Enter ReentranLock to learn how to implement tryAcquire with fair and unjust locks, and analyze both cases together.
Not fair lock
static final class NonfairSync extends Sync {
private static final long serialVersionUID = 7316153563782823691L;
protected final boolean tryAcquire(int acquires) {
returnnonfairTryAcquire(acquires); }}abstract static class Sync extends AbstractQueuedSynchronizer {
@ReservedStackAccess
final boolean nonfairTryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {//state contention lock for the first time
if (compareAndSetState(0, acquires)) { // The state exchange is successful
setExclusiveOwnerThread(current);// Set the property variable
return true; }}else if (current == getExclusiveOwnerThread()) { // Lock reentrant, only state += acquires is handled
int nextc = c + acquires;
if (nextc < 0) / / int crossing the line
throw new Error("Maximum lock count exceeded");
setState(nextc);
return true;
}
return false; }}Copy the codeVery simple, use CAS(exchange comparison) to set the successful thread first, even if it is successful, in determining whether the thread to obtain the lock is holding the lock thread, support reentrant lock.
Fair lock
static final class FairSync extends Sync {
@ReservedStackAccess
protected final boolean tryAcquire(int acquires) {
final Thread current = Thread.currentThread();
int c = getState();
if (c == 0) {
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 codeThe main distinction is made by a HasqueuedToraise judgment, into the analysis of the method
public final boolean hasQueuedPredecessors(a) {
Node h, s;
if((h = head) ! =null) {// The queue is already initialized
CANCELLED, s is CANCELLED when its value is greater than 0. The next node cannot be CANCELLED
if ((s = h.next) == null || s.waitStatus > 0) {
s = null; // traverse in case of concurrent cancellation
for(Node p = tail; p ! = h && p ! =null; p = p.prev) { // Go all the way from the tail chain
if (p.waitStatus <= 0) // The waitStatus status is normals = p; }}if(s ! =null&& s.thread ! = Thread.currentThread())// The thread that acquires the lock as long as it is not the next thread to acquire the lock
return true;
}
return false;
}
Copy the codeCheck whether the head node is initialized. If not, return false. Head already exists, returns true only if the node thread is not equal to the current thread. Why is that? Imagine that when the lock is released and the next node is woken up to acquire the lock, a thread may also acquire the lock, preempting the node in the queue to acquire the lock, which is equivalent to queue-jumping. This is “unfair”. The main difference between public and unfair is that when acquiring a lock, it determines whether there are queuing threads in the queue first. If there are queuing threads, the queue will fail to obtain the lock directly, and the queue will join the queue to ensure that the queue with the longest queue can obtain the lock first. Fair lock will be a little more than unfair lock performance consumption, but the impact is not very big, in the usual development of the use of fair lock is also a good choice, after all, we are choosing to pursue fair 😝.
Release the lock
ReentranLock. Unlock () code
public void unlock(a) {
sync.release(1);
}
Copy the coderelease
public final boolean release(int arg) {
if (tryRelease(arg)) { // This requires a subclass to implement
Node h = head;
if(h ! =null&&stat h.waitStatus ! =0) // The state cannot be 0
unparkSuccessor(h); // Wake up the next queue thread
return true;
}
return false;
}
Copy the codeH.w. aitStatus = 0 is the state or the default state, we know by shouldParkAfterFailedAcquire head state is modified, the threads on the spin lock, also don’t need to wake up.
tryRelease
ReentranLock’s tryRelease is an implementation of fair and unfair lock release.
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
if(Thread.currentThread() ! = getExclusiveOwnerThread())// The thread that releases the lock must be the owner thread, otherwise an exception will be thrown
throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) { // State must be equal to 0 before the lock is released
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
Copy the codeWhenever tryRelease() is called, you can get state-releases without waking up the thread, as long as state=0, and set the owning thread to NULL. Wake up the waiting thread in the queue. A ReentranLock acquiring a lock lock is released process has been covered, but there are many methods in AbstractQueuedSynchronizer, in line with the parse the class, have other functions are also analyzed.
lockInterruptibly
The lockInterruptibly thread can be interrupted to force out of the lock race while acquiring the lock or waiting in a queue. This is a ReentrantLock feature, and the synchronized keyword does not support obtaining lock interrupts. Already the code
public void lockInterruptibly(a) throws InterruptedException {
sync.acquireInterruptibly(1);
}
Copy the codeacquireInterruptibly
AcquireInterruptibly Looks like this
public final void acquireInterruptibly(int arg)
throws InterruptedException {
if (Thread.interrupted()) // It has been interrupted, and the interrupted state is cleared
throw new InterruptedException();
if(! tryAcquire(arg))// Failed to obtain the lock
doAcquireInterruptibly(arg); // Failed to join the column
}
Copy the codeHow does enqueued thread support interrupts
doAcquireInterruptibly
private void doAcquireInterruptibly(int arg)
throws InterruptedException {
final Node node = addWaiter(Node.EXCLUSIVE); // Join the current node of the queue
try {
for (;;) { / / spin
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
return;
}
// Check the state of the front node to remove invalid threads
if (shouldParkAfterFailedAcquire(p, node) &&
parkAndCheckInterrupt()) // The suspended thread returns the thread interrupted state
throw newInterruptedException(); }}catch (Throwable t) {
cancelAcquire(node);
throwt; }}Copy the codeThe lock acquisition process is basically the same as the lock method in that it determines that the thread is interrupted before acquiring the lock and handles it. In the enqueued thread, the thread that has been suspended cannot handle interrupts, as long as the thread is woken up and directly throws an exception to exit the lock contention.
Get lock timeout
ReentrantLock has a way to set the lock to be acquired at a specified time, so that the thread is not blocked while waiting for the lock to be acquired. By setting the timeout, the caller can be left to handle the lock acquisition failure. Get right into the code
public boolean tryLock(long timeout, TimeUnit unit)
throws InterruptedException {
return sync.tryAcquireNanos(1, unit.toNanos(timeout));
}
Copy the codeHow does AQS work
public final boolean tryAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (Thread.interrupted())
throw new InterruptedException();
return tryAcquire(arg) || // Fetching failed enter the following method
doAcquireNanos(arg, nanosTimeout);
}
Copy the codedoAcquireNanos
private boolean doAcquireNanos(int arg, long nanosTimeout)
throws InterruptedException {
if (nanosTimeout <= 0L)
return false;
final long deadline = System.nanoTime() + nanosTimeout; // The timeout timestamp
final Node node = addWaiter(Node.EXCLUSIVE); / / team
try {
for (;;) {
final Node p = node.predecessor();
if (p == head && tryAcquire(arg)) {
setHead(node);
p.next = null; // help GC
return true;
}
nanosTimeout = deadline - System.nanoTime();
if (nanosTimeout <= 0L) { // Timeout has occurred
cancelAcquire(node); // Exit the queue
return false;
}
if (shouldParkAfterFailedAcquire(p, node) &&
nanosTimeout > SPIN_FOR_TIMEOUT_THRESHOLD) // Threads that are larger than 1000 ns need to be suspended otherwise the time is too short to be suspended
LockSupport.parkNanos(this, nanosTimeout); // Suspend the thread for a specified time than wake up
if (Thread.interrupted())
throw newInterruptedException(); }}catch (Throwable t) {
cancelAcquire(node);
throwt; }}Copy the codeAccording to Baidu’s description of nanoseconds, it takes about 2 to 4 nanoseconds for a computer to execute an instruction (such as adding two numbers), too small a time interval to be meaningful. This method has more than the normal method of timeout detection and timeout wake up thread, and everything else.
ConditionObject
ConditionObject AbstractQueuedSynchronizer inner class, realize the Condition the main functions of interface plug threads and awaken jams, similar to wait, notfiy and notifyAll, It is generally used to wake up producers and consumers of thread blocking in synchronous queues. First analyze the Condition interface each method meaning, in the specific analysis method implementation.
Condition
public interface Condition {
// Make the current thread wait until it receives a signal or interrupts
void await(a) throws InterruptedException;
// Make the current thread wait until it receives a signal
void awaitUninterruptibly(a);
// Makes the current thread wait until it receives a signal or exceeds the specified time or emits an interrupt
long awaitNanos(long nanosTimeout) throws InterruptedException;
/ / same as above
boolean await(long time, TimeUnit unit) throws InterruptedException;
// The current thread waits until it receives a message or interrupts or exceeds the specified time
boolean awaitUntil(Date deadline) throws InterruptedException;
// Wake up a single waiting thread
void signal(a);
// Wake up all waiting threads
void signalAll(a);
Copy the codeBefore parsing the implementation class, you need to know the internal properties of ConditionObject
public class ConditionObject implements Condition.java.io.Serializable {
/** First node of condition queue. */
private transient Node firstWaiter;
/** Last node of condition queue. */
private transient Node lastWaiter;
public ConditionObject(a) {}Copy the codeThere are only two internal attributes, the header node and the tail node. Let’s start to analyze the implementation method.
await
public final void await(a) throws InterruptedException {
if (Thread.interrupted()) // return the thread interrupt status and clean up the interrupt signal
throw new InterruptedException();
Node node = addConditionWaiter(); // The thread enters the wait queue
int savedState = fullyRelease(node); // Perform lock release, return state
int interruptMode = 0;
// // Whether the node is in a queue
while(! isOnSyncQueue(node)) {// The thread is not suspended from the queue
LockSupport.park(this);
// When the thread wakes up, it cleans up the interrupted state and rejoins the queue
// When the thread is awakened, it checks whether there is an interrupt signal and returns interruptMode
if((interruptMode = checkInterruptWhileWaiting(node)) ! =0)
break;
}
// Spin the lock in the queue
if(acquireQueued(node, savedState) && interruptMode ! = THROW_IE) interruptMode = REINTERRUPT;if(node.nextWaiter ! =null) // clean up if cancelled
unlinkCancelledWaiters(); // Clear the thread state from CONDITION
if(interruptMode ! =0)
reportInterruptAfterWait(interruptMode);
}
Copy the code- If the thread is interrupted, an exception will be thrown.
- Add Node to the unidirectional list maintained by ConditionObject. This list is used primarily for waking up queues and does not conflict with queues that fetch locks.
- Release lock, at first a little confused here, this should be combined with the synchronization queue to understand. When a thread is suspended, it must release the lock, otherwise it becomes a deadlock. The producer cannot wake up the consumer without obtaining the lock and inserting data.
- The while loop exits as long as the node has joined the thread queue to compete for the lock, or suspends the current thread first. When a thread is awakened, determine if it was awakened because of an interrupt, and if so, jump out of the while loop.
- Now that you are enqueued, you can go and get the lock.
- At this point, the node has acquired the lock. WaitStatus has changed, and it needs to clear the undeclared nodes in the one-way list, including itself.
- When interruptMode is not equal to 0, an interrupt needs to be handled by the caller.
addConditionWaiter
private Node addConditionWaiter(a) {
if(! isHeldExclusively())// Not the current lock owner
throw new IllegalMonitorStateException();
Node t = lastWaiter;
// If lastWaiter is not CONDITION, remove it from the queue. CONDITION is exclusive to the synchronous queue
if(t ! =null&& t.waitStatus ! = Node.CONDITION) { unlinkCancelledWaiters();// Remove status not equal to CONDITION from nextWaiter queue
t = lastWaiter;
}
Node node = new Node(Node.CONDITION); // create a node and set waitStatus to CONDITION
if (t == null) // The head node is not initialized yet
firstWaiter = node;
else
t.nextWaiter = node; // Node is either a head or a tail
lastWaiter = node;
return node;
}
Copy the code
unlinkCancelledWaiters
private void unlinkCancelledWaiters(a) {
Node t = firstWaiter;
Node trail = null;
while(t ! =null) { // Start from the beginning
Node next = t.nextWaiter;
if(t.waitStatus ! = Node.CONDITION) { t.nextWaiter =null; // disconnect t from the following node
if (trail == null) // Delete the header node. The next node is the header node
firstWaiter = next;
else
trail.nextWaiter = next;
if (next == null)
lastWaiter = trail;
}
elsetrail = t; t = next; }}Copy the codeIterate through the condition list and add Node.waitStatus! = node. CONDITION is deleted.
isOnSyncQueue
final boolean isOnSyncQueue(Node node) {
if (node.waitStatus == Node.CONDITION || node.prev == null) // Node has just released the lock and has not entered the queue once
return false;
if(node.next ! =null) // The leading node is not empty and must be in the queue
return true;
return findNodeFromTail(node); // Looking forward from tail to find node returns true
}
Copy the codecheckInterruptWhileWaiting
private int checkInterruptWhileWaiting(Node node) {
return Thread.interrupted() ?
(transferAfterCancelledWait(node) ? THROW_IE : REINTERRUPT) : 0;
}
Copy the codeCheck for thread interruptions and clean it up. No interrupt return 0 directly and no transferAfterCancelledWait execution. Interrupted transferAfterCancelledWait will be executed and returns an interrupt structure.
- REINTERRUPTInterrupt exit while waiting for exit
- THROW_IEThrows an InterruptedException to exit
transferAfterCancelledWait
final boolean transferAfterCancelledWait(Node node) {
if (node.compareAndSetWaitStatus(Node.CONDITION, 0)) { //waitStatus CONDITION => 0 The setting is successful
enq(node); // Join the queue
return true;
}
while(! isOnSyncQueue(node))// Node is not in the queue
Thread.yield(); // The thread abdicates execution and waits for the node to queue up and jump out of its spin
return false;
}
Copy the codeAccording to the previous lock acquisition code, the default value of waitStatus is 0. Only when aitStatus CONDITION => 0 is set successfully can the node join the queue and participate in the lock competition. Failure to set will spin to determine whether the node is in the queue, which is required to exit the method.
reportInterruptAfterWait
private void reportInterruptAfterWait(int interruptMode)
throws InterruptedException {
if (interruptMode == THROW_IE) // Rethrows the interrupt exception to be handled by the caller
throw new InterruptedException();xin
else if (interruptMode == REINTERRUPT) // Interrupt exit while exit
selfInterrupt(a); // The calling thread is interrupted
}
Copy the codeAccording to different interrupt situation, make different processing.
signal
public final void signal(a) {
if(! isHeldExclusively())// The lock must be executed exclusively by the thread
throw new IllegalMonitorStateException();
Node first = firstWaiter;
if(first ! =null) // If the header is not null, the condition queue exists and can be awakened
doSignal(first); // Wake up the thread
}
Copy the codeHow does the wake up thread work
doSignal
private void doSignal(Node first) {
do {
if ( (firstWaiter = first.nextWaiter) == null) // The next node is empty and the list has reached its end
lastWaiter = null;
first.nextWaiter = null;
// This method is the main wake up logic
} while(! transferForSignal(first) && (first = firstWaiter) ! =null);
}
Copy the codetransferForSignal
final boolean transferForSignal(Node node) {
if(! node.compareAndSetWaitStatus(Node.CONDITION,0)) // Join queue waitStatus=0
return false;
// join the queue for the lock
// p is the node prefix
Node p = enq(node);
int ws = p.waitStatus;
// if ws is greater than 0, p will exit the queue and wake up the back node
// p failed to set the state, p node can no longer exist, should wake up the thread to acquire the lock.
if (ws > 0| |! p.compareAndSetWaitStatus(ws, Node.SIGNAL)) LockSupport.unpark(node.thread);// Wake up the node thread
return true; // Wake up successfully returns true
}
Copy the codeThe condition to wake up the thread is whether the front node is available, as long as the current node is ready to be dequeued or has been deleted. Node wakes up to acquire the lock. I’m not going to write any more of these, because it’s too much. Interested in learning for yourself, the content is almost repetitive.
conclusion
This principle by acquiring a lock release lock to parse AbstractQueuedSynchronizer internal source principle, also analyzed the fair and the difference between the fair lock lock. Lock explain some derivative functions, such as lock timeout, interrupt lock, is one of the more full parsing AbstractQueuedSynchronizer as a Java already this similar lock framework supports, ConditionObject is also briefly analyzed to realize thread coordination of signal and await. Most of the content of this article is thought out by myself, if there is something wrong, please come out and discuss with us.