Condition

Before Java5 had Lock, synchronized was used to control synchronization, and wait(), wait(long timeout), notify(), and notifyAll() on Object were used together to realize wait/notification mode. Java5 has seen the advent of the Condition interface, which also provides object-like monitor methods that work with Lock to implement the wait/notification pattern.

Basic method

• Await () : to await the current thread until it receives a signal or is interrupted
• Await (long time, time unit) : to await the current thread until it receives a signal, is interrupted, or reaches the wait time
• Long awaitNanos(Long nanosTimeout) : Keeps the current thread waiting until it receives a signal, is interrupted, or reaches the wait time. The return value represents the time remaining. If nanosTimeout is awakened before nanosTimeout, the return value is nanosTimeout- the time consumed; If the return value is less than 0, a timeout has occurred
• AwaitUninterruptibly () : Causes the current thread to wait until it receives a signal, making it interrupt-insensitive
• The current thread enters the wait state until notified, interrupted, or a specified time is reached. Returns true if wakened without the specified time, false otherwise
• Void signle() : Wakes up a thread waiting on Condition that must acquire the lock associated with Condition before returning from the wait method
• Void signalAll() : Wakes up all threads waiting on Condition that must acquire the lock associated with Condition before returning from the wait method

Condition implementation analysis

ConditionObject is the inner class of AQS, and it makes sense to be the inner class of the synchronizer because Condition requires the lock associated with it. Each Condition contains a wait queue, which is the key to realize the wait/notification function of Condition.

Waiting queue

Await queue is a FIFO queue. Each node in the queue contains a thread reference, which is the thread waiting on the Condition object. If a thread calls await() or await methods, the thread will release the lock, construct the node to join the wait queue and enter the wait state. The nodes waiting for queues reuse the internal Node class of the synchronous queue in AQS.

public class ConditionObject implements Condition.java.io.Serializable {
    private static final long serialVersionUID = 1173984872572414699L;
    /** First node of condition queue. */
    // Wait for the head node in the queue
    private transient Node firstWaiter;
    /** Last node of condition queue. */
    // Wait for the last node in the queue
    private transient Node lastWaiter;
    
    / * * * * / has been eliminated
Copy the code

As shown in the figure, Condition has a reference to the beginning and end nodes, and the new node just needs to point the next of the last node to the new node and update the reference to the last node. Updates and inserts do not use CAS because the thread calling the await method must have obtained the lock object and is thread-safe.

public final void await(a) throws InterruptedException {
    // Determine that the current thread release has been interrupted
    if (Thread.interrupted())
        throw new InterruptedException();
    // Add the current thread to the wait queue
    Node node = addConditionWaiter();
    / / releases the lock
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    // Determine whether the current node is released in the synchronization queue
    // If no, the conditions for competing locks are not met
    // Until the node is detected in the synchronization queue
    while(! isOnSyncQueue(node)) {// Block the current thread
        LockSupport.park(this);
        if((interruptMode = checkInterruptWhileWaiting(node)) ! =0) {
            // If it is interrupted, exit
            break; }}// Spin to get synchronization state/lock
    if(acquireQueued(node, savedState) && interruptMode ! = THROW_IE) interruptMode = REINTERRUPT;// Clear the nodes in the wait queue that are not waiting for the condition
    if(node.nextWaiter ! =null) // clean up if cancelled
        unlinkCancelledWaiters();
    if(interruptMode ! =0)
        reportInterruptAfterWait(interruptMode);
}
Copy the code

private Node addConditionWaiter(a) {
    Node t = lastWaiter;
    // If lastWaiter is cancelled, clean out.
    // Check whether the tail node of the wait queue is not in CONDITION, then clear it
    if(t ! =null&& t.waitStatus ! = Node.CONDITION) { unlinkCancelledWaiters(); t = lastWaiter; } Node node =new Node(Thread.currentThread(), Node.CONDITION);
    if (t == null)
        // The last node is empty, and the current node is treated as the last node
        firstWaiter = node;
    else
        t.nextWaiter = node;// Otherwise, next of the last node points to the current node
    lastWaiter = node;// Assign a reference to the current node to the last node
    return node;
}
Copy the code

notice

Calling Condition’s signal method wakes up the longest waiting node in the waiting queue (the head node) and moves it to the synchronization queue before waking up.

signal

public final void signal(a) {
    if(! isHeldExclusively())// Determine whether the current thread has acquired the lock
        throw new IllegalMonitorStateException();
    // Get the head node
    Node first = firstWaiter;
    if(first ! =null)
        // Wake up the head node
        doSignal(first);
}
Copy the code

private void doSignal(Node first) {
    do {
        // Modify the header node
        if ((firstWaiter = first.nextWaiter) == null)
            lastWaiter = null;
        first.nextWaiter = null;
    } while(! transferForSignal(first) && (first = firstWaiter) ! =null);
}
Copy the code

transferForSignal

final boolean transferForSignal(Node node) {
    /* * If cannot change waitStatus, the node has been cancelled. */
    // The CAS operation changes the current node state from CONDITION to 0
    // If the modification fails, it has been cancelled
    if(! compareAndSetWaitStatus(node, Node.CONDITION,0))
        return false;
    /* * Splice onto queue and try to set waitStatus of predecessor to * indicate that thread is (probably) waiting. If cancelled or * attempt to set waitStatus fails, wake up to resync (in which * case the waitStatus can be transiently and harmlessly wrong). */
    // Call the enq method of AQS to add the current node to the queue
    Node p = enq(node);
    int ws = p.waitStatus;
    // If the node status is CANCEL or fails to change its status to SIGNAL, the node is immediately woken up
    if (ws > 0| |! compareAndSetWaitStatus(p, ws, Node.SIGNAL)) LockSupport.unpark(node.thread);return true;
}
Copy the code

Notification process:

1. Determines whether the current thread has acquired the lock
2. Take the head node out
3. Change the head node in the wait queue to the successor node of the original head node
4. Change the state of the current node from CONDITION to its initial state of 0, return false if the change fails
5. Adds the current node to the synchronization queue
6. Check whether the status of the current node is CANCEL or fails to change its status to SIGNAL

The awakened thread exits from the while loop in the await() method (isOnSyncQueue() returns true) and calls the synchronizer’s acquireQueued method to spin until the lock is acquired. The signalAll method executes the signal method once for each method.