Condition interface for Java concurrent programming

Part of this article is excerpted from The Art of Concurrent Programming in Java

An overview of the

Any Java object has a monitor method, consisting of wait(), wait(long timeout), notify(), and notifyAll(), which work with the synchronized synchronization keyword. The wait-notification pattern can be implemented. The Condition interface also provides object-like monitor methods that, in conjunction with Lock, implement the wait-notification pattern

Object’s monitor method versus the Condition interface:

Interface sample

Condition defines two types of wait-notification methods that require the lock associated with the Condition object to be acquired before the current thread calls these methods. The Condition object is created by the Lock object (which calls the Lock object’s newCondition() method); in other words, the Condition depends on the Lock object

public class ConditionUserCase {

    Lock lock = new ReentrantLock();
    Condition condition = lock.newCondition();

    public void conditionWait(a) throws InterruptedException {
        lock.lock();
        try {
            condition.await();
        } finally{ lock.unlock(); }}public void conditionSignal(a) {
        lock.lock();
        try {
            condition.signal();
        } finally{ lock.unlock(); }}}Copy the code

After calling the await() method, the current thread releases the lock and waits, while another thread calls the signal() method of the Condition object, notifying the current thread, which returns from the await() method and has acquired the lock before returning

Condition partial method and description:

Let’s take a closer look at how Condition is used with an example of a bounded queue

public class BoundedQueue<T> {

    private Object[] items;
    // Subscripts added, subscripts removed, and the current number of data
    private int addIndex, removeIndex, count;
    private Lock lock = new ReentrantLock();
    private Condition notEmpty = lock.newCondition();
    private Condition notFull = lock.newCondition();

    public BoundedQueue(int size) {
        items = new Object[size];
    }

    /** * add an element, if the array is full, add thread to wait until empty */
    public void add(T t) throws InterruptedException {
        lock.lock();
        try {
            while (count == items.length) {
                notFull.await();
            }
            items[addIndex] = t;
            if (++addIndex == items.length) {
                addIndex = 0;
            }
            ++count;
            notEmpty.signal();
        } finally{ lock.unlock(); }}/** * Removes an element from the header. If the array is empty, the deletion thread enters the wait state until a new element is added
    @SuppressWarnings("unchecked")
    public T remove(a) throws InterruptedException {
        lock.lock();
        try {
            while (count == 0) {
                notEmpty.await();
            }
            Object x = items[removeIndex];
            if (++removeIndex == items.length) {
                removeIndex = 0;
            }
            --count;
            notFull.signal();
            return (T) x;
        } finally{ lock.unlock(); }}}Copy the code

Implementation analysis

ConditionObject is synchronizer AbstractQueuedSynchronizer inner classes, each Condition object contains a queue (waiting queue), the queue is the key to the Condition object implementation wait – notifications

1. Wait in a queue

The wait 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 the condition.await () method, the thread releases the lock. Constructs the node, joins the wait queue, and enters the wait state

A Condition contains a wait queue. The Condition has references to the first and last nodes. The new node simply points nextWaiter, the original tail node, to it and updates the tail node. The process of a node referring to an update is not guaranteed with CAS because the thread calling the await() method must be the one that has acquired the lock, i.e. the process is thread-safe by the lock

In the monitor model of Object, an Object has a synchronization queue and a wait queue, while the Lock in the simultaneous sending of packets has a synchronization queue and multiple wait queues, and the corresponding relationship is shown in the figure:

2. Waiting for

Calling the await() method of Condition causes the current thread to enter the wait queue and release the lock while the thread state changes to wait. When returned from await(), the current thread must have acquired the lock associated with Condition

The await() method of Condition looks like this:

public final void await(a) throws InterruptedException {
    // Check thread interrupt status
    if (Thread.interrupted())
        throw new InterruptedException();
    // The current thread is wrapped as Node and queued
    Node node = addConditionWaiter();
    // Release the synchronization state, i.e. release the lock
    int savedState = fullyRelease(node);
    int interruptMode = 0;
    // Check whether the node is in the synchronization queue. If not, wait
    while(! isOnSyncQueue(node)) {// Suspend the thread
        LockSupport.park(this);
        if((interruptMode = checkInterruptWhileWaiting(node)) ! =0)
            break;
    }
    // Contention synchronization state
    if(acquireQueued(node, savedState) && interruptMode ! = THROW_IE) interruptMode = REINTERRUPT;// Clear the node in the condition queue that is not waiting for the condition
    if(node.nextWaiter ! =null) 
        unlinkCancelledWaiters();
    // If the waiting thread is interrupted, an exception is thrown
    if(interruptMode ! =0)
        reportInterruptAfterWait(interruptMode);
}
Copy the code

3. Inform

Calling the signal() method of Condition wakes up the node that has waited the longest in the wait queue (the first node) and moves the node to the synchronization queue before waking it up

Condition’s signal() method code looks like this:

public final void signal(a) {
    // Check whether the current thread has acquired the lock
    if(! isHeldExclusively())throw new IllegalMonitorStateException();
    // Get the first node of the wait queue, move to the synchronization queue and wake up
    Node first = firstWaiter;
    if(first ! =null)
        doSignal(first);
}
Copy the code

Condition’s signAll() method is equivalent to executing a signal() method for each node in the wait queue. The effect is to move all nodes in the wait queue to the synchronization queue and wake up the thread of each node