This article is reprinted from the blog of May’s CangjieThe original
Knowledge of this article
- preface
- What is a AbstractQueuedSynchronizer?
- Implementation principle of ReentrantLock
- Lock () implementation principle
- Unlock () implementation principle
- Implementation of other methods of ReentrantLock
- summary
preface
There are many articles on the Internet about the use of ReentrantLock and the difference between ReentrantLock and synchronized. There are few articles that study ReentrantLock and explain the principle of ReentrantLock clearly. This article will take a look at the implementation principle of ReentrantLock. Research ReentrantLock implementation principle requires a good Java foundation and the ability to read code, some friends do not understand it does not matter, can read later, I believe you will gain. ReentrantLock is implemented based on AQS, as discussed below. The foundation of AQS, in turn, is CAS. For those not familiar with CAS, check out this article on Unsafe and CAS.
What is a AbstractQueuedSynchronizer?
Already implemented is the premise of AbstractQueuedSynchronizer, referred to as “AQS, is Java. Util. The core of concurrent, CountDownLatch, FutureTask, Semaphore, ReentrantLock, and so on all have an inner class that is a subclass of this abstract class. Let’s start with two tables to introduce AQS. The first one is Node, because AQS is based on the realization of FIFO queue, so there must be a Node, Node is a Node. Some properties in Node
| attribute | define |
|---|---|
| Node SHARED = new Node() | Indicates that nodes are in shared mode |
| Node EXCLUSIVE = null | Indicates that Node is in exclusive mode |
| int CANCELLED = 1 | If a Node is set to a cancelled state because of a timeout or interrupt, the cancelled Node should not compete for the lock. It can only remain in the cancelled state and cannot be changed to another state. Nodes in this state are kicked out of the queue and collected by GC |
| int SIGNAL = -1 | Indicates that this Node’s successor is blocked and needs to be notified |
| int CONDITION = -2 | Indicates that the Node is in a condition queue and is blocked waiting for a condition |
| int PROPAGATE = -3 | The head Node may be in this state when used in shared mode, indicating that the next lock acquisition can be propagated unconditionally |
| int waitStatus | 0, the new Node will be in this state |
| Node prev | The precursor Node of a Node in the queue |
| Node next | A successor Node to a Node in the queue |
| Thread thread | The thread held by this Node represents the thread waiting for the lock |
| Node nextWaiter | Represents the next Node waiting for condition |
Having looked at Node, let’s take a look at the variables and methods in AQS:
Some attributes in AQS
| Properties/methods | meaning |
|---|---|
| Thread exclusiveOwnerThread | This is AQS parent AbstractOwnableSynchronizer properties, represent the current owner of the exclusive mode synchronizer |
| Node | The basic unit of a FIFO queue has been described above |
| Node head | The head Node in the FIFO queue |
| Node tail | Tail Node in a FIFO queue |
| int state | Synchronization status: 0 indicates that the lock is not open |
| int getState() | Obtaining synchronization Status |
| setState(int newState) | Setting synchronization Status |
| boolean compareAndSetState(int expect, int update) | CAS is used to set the State |
| long spinForTimeoutThreshold = 1000L | The time the thread spins to wait |
| Node enq(final Node node) | Insert a Node into the FIFO queue |
| Node addWaiter(Node mode) | Creates and expands a wait queue for the current thread and the specified pattern |
| void setHead(Node node) | Sets the head Node of the queue |
| void unparkSuccessor(Node node) | Call up the thread held by Node, if any |
| void doReleaseShared() | Release the lock in shared mode |
| void cancelAcquire(Node node) | Cancels an ongoing Node attempt to acquire a lock |
| boolean shouldParkAfterFailedAcquire(Node pred, Node node) | Whether the current thread should be disabled and wait after a failed attempt to acquire the lock |
| void selfInterrupt() | Interrupts the current thread itself |
| boolean parkAndCheckInterrupt() | Disables the current thread from entering the wait state and interrupts the thread itself |
| boolean acquireQueued(final Node node, int arg) | The thread in the queue acquires the lock |
| tryAcquire(int arg) | Try to get a lockIt is implemented by subclasses of AQS |
| tryRelease(int arg) | Attempt to release the lockIt is implemented by subclasses of AQS |
| isHeldExclusively() | Whether to hold the lock alone |
| acquire(int arg) | Acquiring a lock |
| release(int arg) | Release the lock |
| compareAndSetHead(Node update) | Set the head Node using CAS |
| compareAndSetTail(Node expect, Node update) | Use CAS to set up tail nodes |
| compareAndSetWaitStatus(Node node, int expect, int update) | Use CAS to set the wait state in a Node |
Some of the most important methods and attributes in AQS are listed above. The whole AQS is a typical template mode of the application, very sophisticated design, for the FIFO queue operations have been implemented in AQS, AQS subclasses generally only need to rewrite tryAcquire(int ARg) and tryRelease(int ARG) two methods.
ReentrantLock implementation principle (source more)
ReentrantLock has an abstract class Sync:
private final Sync sync;
/** * Base of synchronization control for this lock. Subclassed * into fair and nonfair versions below. Uses AQS state to * represent the number of holds on the lock. */
abstract static class Sync extends AbstractQueuedSynchronizer {... }Copy the codeReentrantLock instantiates Sync’s implementation classes FairSync and NonfairSync, representing fair and unfair Sync, respectively, based on the Boolean arguments passed into the constructor. Because of ReentrantLock we use a lot of unfair locks, so let’s take a look at how unfair locks are implemented. If thread 1 calls ReentrantLock’s lock() method, thread 1 will monopolize the lock, and the whole call chain is very simple:
- 1, set AbstractQueuedSynchronizer state to 1
- 2, set the AbstractOwnableSynchronizer thread to the current thread
These two steps indicate that thread 1 has exclusive possession of the lock. Thread 2 then tries to acquire the same lock, which cannot be done without thread 1 releasing the lock, so thread 2 blocks. So, how can thread 2 be blocked? Looking at thread 2’s method call chain, it gets more complicated:
Implementation of Lock ()
final void lock(a) {
if (compareAndSetState(0.1))
setExclusiveOwnerThread(Thread.currentThread());
else
acquire(1);
}
Copy the codeThread 1 has set state to 1, so line 2 must be false. Therefore, thread 2 will acquire the acquire method on line 5:
public final void acquire(int arg) {
if(! tryAcquire(arg) && acquireQueued(addWaiter(Node.EXCLUSIVE), arg)) selfInterrupt(); }Copy the codeIf the result is false, the lock fails. If the result is false, the lock fails. If the result is false, the lock fails. So let’s take a look at what the tryAcquire method does, which ends up calling Sync’s nonfairTryAcquire method:
1.final boolean nonfairTryAcquire(int acquires) {
2. final Thread current = Thread.currentThread();
3. int c = getState();
4. if (c == 0) {
5. if (compareAndSetState(0, acquires)) {
6. setExclusiveOwnerThread(current);
7. return true;
8.}9.}10. else if (current == getExclusiveOwnerThread()) {
11. int nextc = c + acquires;
12. if (nextc < 0) // overflow
13. throw new Error("Maximum lock count exceeded");
14. setState(nextc);
15. return true;
16 }
17. return false;
18.}
Copy the codeSince state is volatile, it is visible to thread 2, which takes the latest state and checks again to see if it can hold the lock (thread 1 May have released the lock because its synchronization code was running fast), or returns false.
Note lines 10 to 16 that this code allows a thread to call the same ReentrantLock multiple times, giving state+1 for each call. Since a thread already holds the lock, there is no contention, so there is no need to use CAS to set state (equivalent to a biased lock). If nexTC <0 and an error is raised, the same lock can be reentered at most integer. MAX_VALUE times.
Then we go to the second if method, which uses the addWaiter method of AQS:
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)) {
pred.next = node;
return node;
}
}
enq(node);
return node;
}
Copy the codeCreate a Node for the current thread in exclusive mode (because the mode passed in is NULL). Then check whether there are any nodes in the queue. If there are no nodes in the queue, create a queue.
1.private Node enq(final Node node) {
2. for (;;) {
3. Node t = tail;
4. if (t == null) { // Must initialize
5. if (compareAndSetHead(new Node()))
6. tail = head;
7.}else {
8. node.prev = t;
9. if (compareAndSetTail(t, node)) {
10. t.next = node;
11. return t;
12.}13.}14.}15.}
Copy the codeThis method should be easier to understand if you draw a graph. After forming a queue, it looks like this:
1.final boolean acquireQueued(final Node node, int arg) {
2. boolean failed = true;
3. try {
4. boolean interrupted = false;
5. for (;;) {
6. final Node p = node.predecessor();
7. if (p == head && tryAcquire(arg)) {
8. setHead(node);
9. p.next = null; // help GC
10. failed = false;
11. return interrupted;
12.}13. if (shouldParkAfterFailedAcquire(p, node) &&
14. parkAndCheckInterrupt())
15. interrupted = true;
16.}17.}finally {
18. if (failed)
19. cancelAcquire(node);
20.}21.}
Copy the codeThread 2 is the first Node in the bidirectional queue (with h in front of it), so in line 5 to line 11, thread 2 can acquire the lock again. If thread 1 is still unable to acquire the lock, thread 2 can obtain the lock. First call AQS shouldParkAfterFailedAcquire (p, node) method:
1.private static boolean shouldParkAfterFailedAcquire(Node pred, Node node) {
2. int ws = pred.waitStatus;
3. if (ws == Node.SIGNAL)
4. /* 5. * This node has already set status asking a release 6. * to signal it, so it can safely park. 7. */
9. return true;
10. if (ws > 0) {
11. /* 12. * Predecessor was cancelled. Skip over predecessors and 13. * indicate retry. 14. */
15. do {
16. node.prev = pred = pred.prev;
17.}while (pred.waitStatus > 0);
18. pred.next = node;
19.}else {
20. /* 21. * waitStatus must be 0 or PROPAGATE. Indicate that we 22. * need a signal, but don't park yet. Caller will need to 23. * retry to make sure it cannot acquire before parking. 24. */
25. compareAndSetWaitStatus(pred, ws, Node.SIGNAL);
26.}27. return false;
28.}
Copy the codeThis waitStatus is the waitStatus of h, which is obviously 0, so set H’s waitStatus to noed. SIGNAL (-1) and return false. Since return false, the above acquireQueued of line 13 if nature, go again for loop, or try to acquiring a lock, you don’t succeed, keep going shouldParkAfterFailedAcquire, waitStatus at this time of 1, 3 lines of judgment, Returns true. Then go to the second judgment condition parkAndCheckInterrupt on line 13 of the acquireQueued if:
private final boolean parkAndCheckInterrupt(a) {
LockSupport.park(this);
return Thread.interrupted();
}
Copy the codepublic static void park(Object blocker) {
Thread t = Thread.currentThread();
setBlocker(t, blocker);
UNSAFE.park(false.0L);
setBlocker(t, null);
}
Copy the codeAs a final step, the park method that calls LockSupport (the blocking primitive) blocks the current thread. At this point, the complete process of using ReentrantLock to let thread 1 monopolize the lock and thread 2 enter the FIFO queue and block is complete.
Now that we know what to do with lock(), it’s time to explore what the code does with unlock(), and move on to the next section.
The implementation principle of UNLOCK
ReentrantLock: Unlock ();
public void unlock(a) {
sync.release(1);
}
Copy the codeGo AQS release:
1.public final boolean release(int arg) {
2. if (tryRelease(arg)) {
3. Node h = head;
4. if(h ! =null&& h.waitStatus ! =0)
5. unparkSuccessor(h);
6. return true;
7.}8. return false;
9.}
Copy the codeCall Sync’s tryRelease to try to release the lock:
protected final boolean tryRelease(int releases) {
int c = getState() - releases;
if(Thread.currentThread() ! = getExclusiveOwnerThread())throw new IllegalMonitorStateException();
boolean free = false;
if (c == 0) {
free = true;
setExclusiveOwnerThread(null);
}
setState(c);
return free;
}
Copy the codeFirst, free=true is only allowed if c==0. This corresponds to the number of times a thread calls lock() and must call unlock() the same number of times.
When a thread for the same after all already unlocked, AQS state of nature is 0, the AbstractOwnableSynchronizer exclusiveOwnerThread will be set to null, it means no thread has a lock, Method returns true. If the code goes further, at line 4 of the release method above, h is not null and waitStatus of H is -1 which is not equal to 0, so go to the unparkprecursor method at line 5:
1.private void unparkSuccessor(Node node) {
2. int ws = node.waitStatus;
3. if (ws < 0)
4. compareAndSetWaitStatus(node, ws, 0);
5. Node s = node.next;
6. if (s == null || s.waitStatus > 0) {
7. s = null;
8. for(Node t = tail; t ! =null&& t ! = node; t = t.prev)9. if (t.waitStatus <= 0)
10. s = t;
11.}12. if(s ! =null)
13. LockSupport.unpark(s.thread);
14.}
Copy the codeS is the next Node of h, and the thread in this Node is thread 2. Since this Node is not null, thread 2 will be unPark after 12 lines and run. An important question is: how do I reduce the number of nodes in the FIFO queue when the lock is unlocked? This is a clever design that goes back to the acquireQueued method of AQS:
1.final boolean acquireQueued(final Node node, int arg) {
2. boolean failed = true;
3. try {
4. boolean interrupted = false;
5. for (;;) {
6. final Node p = node.predecessor();
7. if (p == head && tryAcquire(arg)) {
8. setHead(node);
9. p.next = null; // help GC
10. failed = false;
11. return interrupted;
12.}13. if (shouldParkAfterFailedAcquire(p, node) &&
14. parkAndCheckInterrupt())
15. interrupted = true;
16.}17.}finally {
18. if (failed)
19. cancelAcquire(node);
20.}21.}
Copy the codeThe blocked thread 2 is blocked on line 14, notice that there is no return statement, that is, thread 2 will still do the for loop after completing the block. Thread 2 tries tryAcquire and succeeds. Then thread 2 becomes the head Node. Set p’s next to null, so that all objects in the original head Node do not point to any block of memory. The end of a method is automatically reclaimed, so that the original head Node has no references to it after the method is called, and it is automatically reclaimed by the GC. At this point, a return statement is encountered, and the acquireQueued method ends. The following nodes work the same way.
Here’s a detail. Take a look at the setHead method:
private void setHead(Node node) {
head = node;
node.thread = null;
node.prev = null;
}
Copy the codeThe setHead method has Null precursor nodes and no threads, so why not use a waiting thread as the Head Node?
Since a thread can be cancelled at any time due to an interrupt, the Node will naturally be GC, and the subsequent Node of the first Node will have to be a new head before GC, which can be very troublesome. Using a Node with no thread as a head acts as a bootstrap, since the head has no threads and cannot be cancelled.
Just to prevent the next node of head from being cancelled, go through the end to find the node nearest the head and unPark that node.
Implementation of other methods of ReentrantLock
If you understand how ReentrantLock is implemented, you will find that the rest of the methods in ReentrantLock are fairly simple to implement.
- 1, int getHoldCount ()
final int getHoldCount(a) {
return isHeldExclusively() ? getState() : 0;
}
Copy the codeThe lock() method that gets the ReentrantLock is called several times, which is the current value of state
- 2, Thread getOwner ()
final Thread getOwner(a) {
return getState() == 0 ? null : getExclusiveOwnerThread();
}
Copy the codeGet the current thread holds the lock, is the value of the AbstractOwnableSynchronizer exclusiveOwnerThread
- 3, the Collection getQueuedThreads ()
public final Collection<Thread> getQueuedThreads(a) {
ArrayList<Thread> list = new ArrayList<Thread>();
for(Node p = tail; p ! =null; p = p.prev) {
Thread t = p.thread;
if(t ! =null)
list.add(t);
}
return list;
}
Copy the codeLet’s go through it from end to end, add it to the ArrayList
- 4, int getQueuedLength ()
public final int getQueueLength(a) {
int n = 0;
for(Node p = tail; p ! =null; p = p.prev) {
if(p.thread ! =null)
++n;
}
return n;
}
Copy the codeSo let’s go all the way from end to end, plus n. Of course, this method and the above method may not be accurate, because other threads may add nodes to the end of the queue while traversing.
Other methods are also about the same, you can go to see.
summary
In this paper, from the source code analysis of ReentrantLock unfair lock and lock release principle, through the source code we know that the cornerstone of ReentrantLock is AQS, AQS is the cornerstone of Node, multi-threaded lock competition through the spin waiting for resources, Unsafe provides the compareAndSwapObject method to ensure that only one thread can acquire the lock when multiple threads compete for the lock, thereby ensuring thread safety. Look at the fair lock source, each time in the acquisition of the lock to check whether the FIFO queue has a waiting thread, if there is to put their own additional to the end of the FIFO queue, and not fair lock is to try to get the lock every time, not to join the FIFO queue. In addition, the code in this article is not copied from the original blogger’s code, but in jdK8 source code based on the copy over, and the original blogger posted a little deviation from the code, but the core implementation is consistent.