JUC concurrent programming 09 -- source code analysis of condition implementation

have a look ReentrantLock Medium newCondition Method

final ConditionObject newCondition() {
    
      return new ConditionObject();
}

Then click in and find that it is AQS Inner class in .

 public class ConditionObject implements Condition, java.io.Serializable {
    
        private static final long serialVersionUID = 1173984872572414699L;
        private transient Node firstWaiter;
        private transient Node lastWaiter;
        .....
}

Found no , There is also the definition of linked list operation , The principle of its implementation is conditional queue . Its structure can refer to the following figure .

We know , When a thread calls await When the method is used , Will enter the waiting state , Until used by other threads signal Method wake up . The waiting queue here is used to store the data in await The number of threads waiting .

Let's take a look at the most critical await How are methods implemented . Here, let's start with a few premises .

• Only threads that already hold locks can call this method .
• When this method is called , Will release the lock directly , No matter how many locks are added .
• Only other threads call signal Method or interrupt will wake up the waiting thread .
• After being awakened, wait for other threads to release the lock , Only after you get the lock can you continue to execute , And it will return to its previous state (await After adding several layers of locks before, there are still several layers of locks ).

Take a look await The source code of the method .

       public final void await() throws InterruptedException {
    
            if (Thread.interrupted())
                throw new InterruptedException(); // If you're calling await If it has been interrupted before , Throw an interrupt exception directly .
            Node node = addConditionWaiter(); //  Create a new node for the current thread , Add it to the condition queue 
            int savedState = fullyRelease(node); //  Completely release the lock of the current thread , And save the status of the current thread , Because I have to recover 
            int interruptMode = 0; //  Interrupt state 
            while (!isOnSyncQueue(node)) {
     // Loop to determine whether it exists in AQS In the synchronization queue of ( If the waiting thread is awakened by another , Will enter AQS In the synchronization queue of , We will talk later )
                LockSupport.park(this); //  If it's still waiting , Suspend the thread 
                if ((interruptMode = checkInterruptWhileWaiting(node)) != 0) //  Check if it was interrupted while waiting 
                    break;
            }
            //  Jumping out of the loop means that the thread must have been awakened , At this time, you only need to get the lock to run 
            if (acquireQueued(node, savedState) && interruptMode != THROW_IE) //  Just try to get the lock , The process here is basically the same as that of a thread grabbing a lock 
                interruptMode = REINTERRUPT; 
             //  After getting the lock, you can basically continue to run , Here are some basic cleaning work 
            if (node.nextWaiter != null) // clean up if cancelled
                unlinkCancelledWaiters();
            if (interruptMode != 0) //  It's still a response interrupt 
                reportInterruptAfterWait(interruptMode);
        }

The above process is very regular , Nothing more than responding to interrupts 、 Thread hanging 、 Strong lock 、 Basic logic such as cleaning up , Look again. signal How is the method implemented . Before reading the source code , Let's clarify a few points first :

• Only the thread holding the lock can call siganl Method to wake up other threads .
• First in the priority wake-up condition queue , If there is a problem during wake-up , Then look down , Until you find the first thread that can be awakened .
• In essence, the wake-up result is to throw the nodes of the conditional queue directly into the waiting queue , Let it compete for lock resources .
• After getting the lock , The thread can continue to execute .

The process can refer to the following figure .
Source code .

        public final void signal() {
    
            if (!isHeldExclusively()) //  Check whether the current thread holds the lock 
                throw new IllegalMonitorStateException(); //  You cannot wake up other threads without holding a lock 
            Node first = firstWaiter; //  Get the first node of the condition queue 
            if (first != null) // The condition queue is not empty 
                doSignal(first); //  Wake up the 
        }

doSignal

   private void doSignal(Node first) {
    
            do {
    
                if ( (firstWaiter = first.nextWaiter) == null) //  If the current node has no successor nodes , The condition queue is directly empty 
                    lastWaiter = null; //  This is equivalent to clearing the condition queue 
                first.nextWaiter = null; //  Set the... Of the current node nextWaiter Set to empty , This is because the current node is out of the conditional queue 
            } while (!transferForSignal(first) && //  Wake up the first node , There is no awakening ( Be cancelled ) And if the condition queue is not empty, it will cycle all the time 
                     (first = firstWaiter) != null);
        }

transferForSignal

    final boolean transferForSignal(Node node) {
    
        //  If it's here CAS Failure , Probably because the thread was canceled 
        if (!compareAndSetWaitStatus(node, Node.CONDITION, 0))
            return false;
            
        // CAS success , Drop the node directly into AQS Waiting in the queue 
        Node p = enq(node); // enq Returns the predecessor node of the incoming node , I don't remember. I can look down 
        //  End of core logic , The following code is a layer of optimization , You can jump out of the previous while loop 
        int ws = p.waitStatus; //  Save the waiting state of the precursor node 
        //  If the status of the precursor node is cancel , perhaps CAS Set it to signal Failure ( May be in ws>0 The moment after the cancellation )
        if (ws > 0 || !compareAndSetWaitStatus(p, ws, Node.SIGNAL)) 
            LockSupport.unpark(node.thread); //  Wake up the thread directly 
        return true;
    }

I don't know how to join the team. Do you remember . I don't remember here. You can look at the code .

    private Node enq(final Node node) {
    
        for (;;) {
    
            Node t = tail;
            if (t == null) {
     
                if (compareAndSetHead(new Node()))
                    tail = head;
            } else {
    
                node.prev = t;
                if (compareAndSetTail(t, node)) {
    
                    t.next = node;
                    return t;
                }
            }
        }
    }

The figure below summarizes await,signal The fault of the .