Concurrency: How to Wait

Basic wait

When we want to coordinate multiple threads to run in order, there are many useful synchronizer classes in Java library for us to use, like Semaphore, CyclicBarrier, CountDownLatch etc.

A common usage of synchronizer class is consumer and producer problem. If we decide to use some basic method from internal lock, we can do it like following:

// thread one
while (ready) {
    obj.wait();
}

// thread two
ready = true;
obj.notify(); // or, depends on situation, obj.notifyAll();

Or we can use explicit lock: Lock & Condition to do similar thing.

condition.await();
condition.signal();

And the above code is more a practice than a real usage. If you really need to write a producer and consumer solution, BlockingQueue may be more convenient to do it.

waitForAny

When we get familiar with the common use of wait, we are going to implement waitForAny utility function based on other library code of java.

Definition

As what we always do, we have to define our problem before to solve it. Here we are going to implement the function with following signature:

/**
 * wait until any object invoke notify/notifyAll
 * @return index of which object that was notified
 */
int waitForAny(Object[] objs) {
}

Simple thought

The simple thought is the translation of the function name: let each object wait on a thread, and if any one is notified, it can cancel all waiting and return from this method.
But it is easier said than done. We have a lot of details to cover:

• First, we have to let the caller thread wait.
• Second, we have to let each object wait on separate thread.
• Next, we notify the caller thread to run.
• Finally, caller thread cancel all waiting thread when any object is notified.

And the first and second steps should be atomic operation, or we may miss some notification before we starting with waiting. So we have to choose a giant lock to protect waitForAny and notify. For the convenience, we just choose the class which contain this method as the lock.
Then we have the following method skeleton:

public synchronized int waitForAny() {
    // finish step one -> four
    return 0;
}

public synchronized void notify(Object o) {
    o.notify();
}

We can use executors to run multiple jobs at the same time, and we wait on each object’s lock in separated thread like following code:

for (int i = 0; i < objects.length; i++) {
    int finalI = i;
    service.submit(() -> {
        synchronized (objects[finalI]) {
            objects[finalI].wait();
        }
        // (1) notify caller here
        return null;
    });
}
// (2) caller wait here

// cancel other waiting to release thread
for (Object object : objects) {
    notify(object);
}        

What should be noticed is that we should not use Executors.newFixedThreadPool() to run waiting jobs, because if more jobs than thread, some jobs will not start until others are finished. And if the object that being notified doesn’t start waiting, we will be in deadlock, waiting forever.

Now, we almost finish the code, we only need a synchronizer class to wait and notify. We have a lot of choices here (Object, Semaphore etc. ), but we just show some of them.

CountDownLatch

This is so handy to handle this case, you can think it as the latch to stop caller. When any object end the waiting, you release the latch, and caller is free to continue.

CountDownLatch latch = new CountDownLatch(1);

// (1) notify caller here
latch.countDown();

// (2) caller wait here
latch.await();

AQS

The AQS is the abbreviation of AbstractQueuedSynchronizer, which is the class used to implement many synchronizer class in Java. So we can definitely use it to finish the job. You can try to understand how AQS works and define you customized synchronizer.

invokeAny

We finish the job using synchronizer to get familiar the library class. But for this problem, executor has more clear way to do it:

List<Callable<Void>> jobs = new ArrayList<>(objects.length);
for (Object object : objects) {
    jobs.add(() -> {
        synchronized (object) {
            object.wait();
        }
        return null;
    });
}
service.invokeAny(jobs);

Because invokeAny is blocked until any job is finished and it will cancel other automatically, which just what we need, this code is much more clear.

Practice

How to implement waitForAll(i.e. wait until all objects invoke notify)?

Some trials can be found in my git repo and welcome your advice.

Discussion

Do you have any other ideas? Share us in comment.

Written with StackEdit.