From the school to A factory all the way sunshine vicissitudes of life

Please go to www.codercc.com

1. The introduction of ConcurrentLinkedQueue

Collection classes, such as ArrayList and HashMap, are often used in single-threaded programming, but they are not thread-safe. There are also a lot of points to check in an interview, such as ArrayList is not thread-safe, Vector is thread-safe. The way to ensure Vector thread safety is to serialize multithreaded execution by using synchronized exclusive locks on methods in a crude manner. ArrayList to be thread-safe can also use the Collections. SynchronizedList ArrayList < T > List (List) method into thread-safe, However, this conversion mode is still achieved through synchronized modification, which is obviously not an efficient way. Meanwhile, queues are also a data structure commonly used by us. In order to solve the problem of thread safety, ConcurrentLinkedQueue Is a thread-safe queue prepared by Doug Lea. As you can tell from the class name, the data structure that implements the queue is a chain.

1.1 the Node

To learn about ConcurrentLinkedQueue, you have to start with its node class and understand its underlying data structure. The Node class source code is:

private static class Node<E> { volatile E item; volatile Node<E> next; . }Copy the code

The Node Node contains two fields: the item field and the next pointer, which points to the next Node to form a chained queue. And they are all decorated with volatile to ensure memory visibility (see this article on volatile). ConcurrentLinkedQueue contains two member variables:

private transient volatile Node<E> head;
private transient volatile Node<E> tail;
Copy the code

ConcurrentLinkedQueue manages queues by holding Pointers to the start and end of the queue. When we call the no-argument constructor, its source is:

public ConcurrentLinkedQueue() {
    head = tail = new Node<E>(null);
}
Copy the code

The head and tail Pointers point to a node whose item field is null, and ConcurrentLinkedQueue looks like this:

In the figure, head and tail point to the same node Node0, where the item field is null and the next field is null.

1.2 Performing CAS Operations on a Node

When the queue is in the queue, it is inevitable to operate the node, and it is easy to appear the problem of thread safety in multithreading. As you can see, once the processor instruction set supports the CMPXCHG instruction, CAS operations are used in Java source code for concurrent processing (see section 3.1 of this article for CAS operations). CAS operations on nodes in ConcurrentLinkedQueue

/ / change the Node in the data domain item Boolean casItem (E, E val CMP) {return UNSAFE.com pareAndSwapObject (this, itemOffset, CMP, val); Next void lazySetNext(Node<E> val) {unsafe. putOrderedObject(this, nextOffset, val); } / / change the Node pointer field in the next Boolean casNext (Node < E > CMP, Node < E > val) {return UNSAFE.com pareAndSwapObject (this, nextOffset, cmp, val); }Copy the code

UNSAFE is a class called sun.misc. UNSAFE, which is the underlying method for hotspot. The UNSAFE class provides CAS operations after all.

2. Offer method

For a queue, inserts satisfy the FIFO property that insert elements are always inserted at the end of the queue, while fetch (remove) elements are always inserted from the head of the queue. The natural place to start to thoroughly understand ConcurrentLinkedQueue is with the Offer and poll methods. To understand the Offer method, debug the code line by line. In addition, when looking at multithreaded code, think of it this way:

Poll —-offer is faster than poll ——– the queue length will get longer and longer, because the offer node is always at the end of the queue, The poll node is always at the head of the queue, which means that there is no “intersection” between offer thread and poll thread, which means that the two threads do not affect each other. From the perspective of relative speed, —- is slower than poll ——– the relative rate of poll is faster than offer, that is, the head of the queue is removed faster than the end of the queue is added faster. As a result, the queue length gets shorter and shorter. Offer thread and poll thread will appear “intersection”, that is, at that moment, it can be called the critical point of the node where offer thread and poll thread operate simultaneously, and the offer thread and poll thread must influence each other at this node. According to the relative order of offer and poll at the critical point, it can be considered from two perspectives: 1. The order of execution is offer >poll >offer, which indicates that when the offer thread inserts Node2 after Node1, the poll thread has deleted Node1. 2. Poll –>offer–>poll, i.e. when the poll thread is ready to delete the node (the queue is empty), the offer thread inserts a node to make the queue become non-empty

Let’s start with this code:

1. ConcurrentLinkedQueue<Integer> queue = new ConcurrentLinkedQueue<>();
2. queue.offer(1);
3. queue.offer(2);
Copy the code

Create an instance of ConcurrentLinkedQueue, offer 1 and then offer 2. Source code of offer:

public boolean offer(E e) { 1. checkNotNull(e); 2. final Node<E> newNode = new Node<E>(e); 3. for (Node<E> t = tail, p = t;;) { 4. Node<E> q = p.next; 5. if (q == null) { 6. // p is last node 7. if (p.casNext(null, newNode)) { // Successful CAS is the linearization point // for e to become an element of this queue, // and for newNode to become "live". 8. if (p ! = t) // hop two nodes at a time 9. casTail(t, newNode); // Failure is OK. 10. return true; } // Lost CAS race to another thread; re-read next } 11. else if (p == q) // We have fallen off list. If tail is unchanged, it // will also be off-list, in which case we need to // jump to head, from which all live nodes are always // reachable. Else the new tail is a better bet. 12. p = (t ! = (t = tail)) ? t : head; else // Check for tail updates after two hops. 13. p = (p ! = t && t ! = (t = tail)) ? t : q; }}Copy the code

Single-thread execution Angle analysis:

Start by analyzing the process of Offer 1 from the perspective of single-threaded execution. Line 2 wraps e as a Node class, and line 3 performs a for loop with only the initialization condition and no loop termination condition. This is very consistent with the CAS “routine”. In the body of the loop, a successful CAS operation returns directly, and if the CAS operation fails, the for loop continues to retry until it succeeds. Here the instance variable t is initialized to tail and p is initialized to t (tail). For the sake of understanding, p is considered the true tail of the queue. Tail does not necessarily refer to the true tail of the object, because in ConcurrentLinkedQueue tail is updated late. As the code goes to line 3, t and p both point to Node0, whose item field is null and next field is null, respectively, created during initialization. In line 4, variable q is assigned null, and in line 5, if is judged to be true. In line 7, casNext is used to set the inserted Node as the next Node of the Node P at the end of the current queue. If the CAS operation fails, the loop ends and retry in the next loop. The CAS operation goes to line 8 successfully. At this point, p==t, if false, return true. If 1 is inserted successfully, ConcurrentLinkedQueue will look like the following:

As shown in the figure, the tail node of the queue should be Node1, and the node that tail points to is still Node0. Therefore, tail is delayed updating. So let’s move on to offer 2. It’s clear that the node q points to is not null anymore, but Node1. If is false on line 5, if is false on line 11, and the code goes to line 13. Ok, so when we insert the node again we ask ourselves the question, okay? Tail does not point to the tail node of the queue, so the first thing we should do is to find where the current tail node of the queue is inserted. So line 13 is to find the actual tail node of the queue.

Locate the real end node of the queue

p = (p ! = t && t ! = (t = tail)) ? t : q;Copy the code

Let’s look at this line of code, if this code is inSingle-threaded environmentWhen I do that, obviously since p is equal to t, p is going to be assigned q, and q is going to be equal toNode<E> q = p.next, namely the Node1. In the first loop, p points to Node1, the real end node of the queue, so in the next loop, line 4 q points to null. So in line 5 if is true, then in line 7 casNext is still used to set next of p Node as the newly added Node. Then in line 8, p! =t, line 8 if true, passescasTail(t, newNode)Set the current Node Node as the last Node of the queue, and the queue status diagram is as follows:

The node pointed to by tail changes from Node0 to Node2The reason why casTail fails here and does not require retries is that the offer code is mainly passed through p’s next node Q (Node<E> q = p.nextThe logic that determines what happens when casTail fails looks like this:

If casTail fails to set tail to the Node0 node, all it does is loop through 13 lines of code several times to locate the node at the end of the queue.

By analyzing the single-thread execution Angle, we can learn that the execution logic of poll is as follows:

1. If the next node (next domain) of the node pointed to by tail is null, it indicates that the node pointed to by tail is the real node at the end of the queue. Therefore, casNext can be used to insert the current node to be inserted, but tail does not change at this time, as shown in Figure 2.

2. If the next node (the next domain) of the node to which tail points is not null, then the node to which tail points is not the real end of the queue. Step forward through the q (Node
   
   q = p.ext) pointer to find the last Node in the queue, insert the current Node through casNext, and change tail through casTail, as shown in Figure 3.
   

Let’s go back and see that p is equal to p factorial. = t && t ! = (t = tail)) ? t : q; This line of code is in a single thread, this code will never assign p to t, so writing it this way won’t do any good, so let’s try to analyze it in a multithreaded case.

Multi-threaded execution Angle analysis

Multiple thread offer

Obviously there’s more to it than that, and it’s actually a very interesting line of code in a multithreaded environment. t ! = (t = tail) This operation is not an atomic operation.

As shown in the figure, thread A reads variable T at this time, thread B just offers A Node at this time, and then changes the tail pointer. Then thread A executes t=tail again and t points to another Node. It is obvious that the variable t read by thread A is pointing to A different node, namely, t! = (t = tail) is true, and since t points to a change in the node p! = t is also true, in which case the result of this line of code is p and t. The latest pointer to t points to the same node, and t is also the true end of the queue. Now that you’ve located the actual end-of-queue node of the queue, you can perform the Offer operation.

offer->poll->offer

That leaves line 11, which we haven’t analyzed, and presumably answers to a partial thread offer and a partial poll. When if (p == q) is true, it means that the next of the node pointed to by P also points to itself. Such node is called sentinel node, which has little value in the queue and is generally represented as the node to be deleted or empty node. To get a good idea of this, let’s take a look at the poll method and then come back to it, which is interesting :).

3. The poll method

The poll method is as follows:

public E poll() {
    restartFromHead:
    1. for (;;) {
    2.    for (Node<E> h = head, p = h, q;;) {
    3.        E item = p.item;
4. if (item ! = null &amp; &amp; p.casItem(item, null)) { // Successful CAS is the linearization point // for item to be removed from this queue. 5. if (p ! = h) // hop two nodes at a time 6. updateHead(h, ((q = p.next) ! = null) ? q : p); 7. return item; } 8. else if ((q = p.next) == null) { 9. updateHead(h, p); 10. return null; } 11. else if (p == q) 12. continue restartFromHead; else 13. p = q; }}Copy the code
Copy the code
}



Let’s first clarify the basic logic of this approach from the perspective of a single thread. Assume the initial state of ConcurrentLinkedQueue as shown below:

    
In the definition of the offer parameter, we still take the variable P as the queue first to delete the real queue head node. The node pointed to by h (head) is not necessarily the queue head node. P =h=head; p=h=head; p=h=head; Set Node1’s data field to null using casItem. If the CAS fails to be configured, the loop ends and the next loop tries again. If the execution of line 4 is successful and the code enters line 5, both P and h point to Node1. If is judged to be false in line 5, and then directly return to data field 1 of Node1 in line 7. The method is finished, and the queue status at this time is shown as follows.

    
The first thing to do is to poll Node1 (Node1, Node1, Node1, Node1, Node1, Node1, Node1, Node1).

Locate the deleted header node

If (q = p.ext) is also false. Since q refers to Node2, if (q = p.ext) is also false on line 11. So the code goes to line 13, where p and Q together point to Node2, and you find the actual header node to delete. It can be concluded that the process of locating the head node to be deleted is as follows: If the data field of the current node is null, it is obvious that the node is not the node to be deleted, and the next node of the current node is used to test it. After the first cycle, the state is as follows:

    
For the next loop, line 4 performs the same operation as above. Assume that casItem is set successfully in line 4, since P already points to Node2 and H still points to Node1, if in line 5 is judged to be true. Then execute updateHead(h, ((q = p.ext)! = null) ? Q: p), where q points to Node3. All the references passed to updateHead are h references to Node1 and Q references to Node3. UpdateHead ();
final void updateHead(Node<E> h, Node<E> p) { if (h ! = p && casHead(h, p)) h.lazySetNext(h); }Copy the code
This method mainly points the head of the queue to Node3 via casHead and points the next field of Node1 to itself via H.lazysetNext. Finally, line 7 returns the value of Node2. The queue status is shown in the figure below:

    
Node1’s next field points to itself, and head points to Node3. If the queue is empty, line 8 (q = p.ext) == null is executed, if is true, so null is returned on line 10. The above analysis is from the perspective of single thread execution and can also let us know the overall idea of poll. Now let’s make a summary:


If the Item of the current node pointed to by head,h, and P is not null, it indicates that the node is the real head node (node to be deleted). Simply set the Item field to null by casItem method and return the original Item directly.


If the item of the node pointed to by head,h, and P is null, then the node is not the node to be deleted. Probe by making q point to the next node of P (q = p.ext), and if found, update the node pointed to by head with updateHead method and construct sentinel node (h.lazysetNext (h) of updateHead method).


Next, in accordance with the above way of thinking to analyze offer, let’s analyze the situation of multi-threading, the first case is;
Multithreading execution analysis:

Multi-thread poll

Now looking back at the source of the poll method, there is a section like this:
else if (p == q)
    continue restartFromHead;
Copy the code
Q = p next, that is, Q always points to the next node of P. Under what circumstances can P and Q point to the same node? According to our analysis above, only the nodes pointed to P become sentinel nodes when polling (via H. lazysetNext in the updateHead method). When thread A decides that p==q, thread B has completed the poll method to convert the node pointing to P to the sentinel node and the node pointing to head has been changed, so it needs to execute from restartFromHead to ensure that the latest head is used.

poll->offer->poll

If the current queue is empty, thread A poll, thread B poll, and thread A poll, does thread A return null or the latest node that thread B just inserted? Let’s write a generation demo:
public static void main(String[] args) { Thread thread1 = new Thread(() -> { Integer value = queue.poll(); System.out.println(thread.currentThread ().getName() + "poll:" + value); System.out.println("queue is currently empty: "+ queue.isempty ()); }); thread1.start(); Thread thread2 = new Thread(() -> { queue.offer(1); }); thread2.start(); }Copy the code
The output is:

The value of thread-0 poll is: null queue Whether the current poll is empty: false

If ((q = p.ext) == null); if (q = p.ext) == null); Thread1 is paused, and thread2 inserts an offer into a node whose value is 1. Thread1 does not retry. Instead, the code continues down the queue, returning NULL, even though thread2 has inserted a new node of value 1. So the output thread0 poll is null, and the queue is not empty. Therefore, whether a queue isEmpty or not cannot be determined by returning null in poll time. IsEmpty can be used to determine whether a queue isEmpty.
4. Some threads in the offer method poll some threads
One problem we were left with while analyzing the Offer method was understanding line 11 of the offer method.

offer->poll->offer

Line 11 of the offer methodif (p == q), which can make if judge true as the node pointed to by PThe sentinel nodeAnd when is the sentinel node constructed? In the discussion of the poll method, we have found the answer: ** When the item field of the node pointed to by head is null, the head is updated after the node to be inserted, and the node pointed to by head is set as the sentinel node. ** Assume that the initial state of the queue looks like the figure below:So when thread A performs offer and thread B performs poll, one of the following situations occurs:
As shown in the figure, the tail node of thread A exists in the next node Node1, so the real tail node of the queue will be searched forward by referring to Q. When the poll operation is performed to judge if (p == q), thread B will perform poll operation at this time. Head and P point to Node0. Since Node0’s item field is null, Node1, the real head node of the queue, will also be found step by step. After thread B completes polling, Node0 will be converted into a sentinel node. This means that the head of the queue has changed, and the queue state is shown below.

    
If (p == q) = true, thread A will continue executing p = (t! = (t = tail)) ? t : head; Since the tail pointer has not changed, p is assigned head, and the insert operation is completed from head again.
5. Design of HOPS
Based on the above analysis of offer and poll methods, we find that tail and head are delayed updates, and their update trigger timing is:
Tail update trigger time: When the next node of the node pointed to by tail is not null, the real tail node of the queue is located. After the tail node is found and inserted, the tail update is performed through casTail. When the next node of the node to which tail points is null, only the node is inserted and tail is not updated.
** When the head is triggered: ** When the item field of the node to which the head points is null, the actual head node of the queue will be located, and the head node will be deleted only after the head node is found. If the item field of the node to which the head points is not null, delete the node and do not update the head.
In addition, during the update operation, the source code will have a comment: Hop two nodes at a time. Therefore, this delayed update strategy is called HOPS for this reason. As can be seen from the above update status graph, the update of head and tail is “hop”, that is, there is always an interval between them. So what is the intent of this design?
If tail is always the last node of the queue, the implementation is less code and the logic is easier to understand. However, there is a disadvantage to doing this. ** If a large number of enqueueing operations are executed each time the CAS is tail updated, it can add up to a significant performance loss. If the operation of CAS update can be reduced, the operation efficiency of joining the team can be greatly improved. Therefore, Doug Lea uses CAS to update tail every time (the distance between tail and the node at the end of the team is 1). ** The same goes for the head update. Although this design increases the number of tail nodes in the loop, overall the read operation is much more efficient than the write performance, so the performance loss of the extra tail node in the loop is relatively small.

The resources

The Art of Concurrent Programming in Java
Java High Concurrency Programming
ConcurrentLinkedQueue blog: https://www.cnblogs.com/sunshine-2015/p/6067709.html