Signaling Threads with Condition Variables

Since you just were introduced to mutexes, now is the time to introduce condition variables. They don’t solve the same problem, but they’re inherently married to mutexes.

So what problem do they solve?

A ConditionVariable can be used to signal one (or many) threads when some event happens, or some state changes, whereas mutexes are a means of synchronizing access to resources. Condition variables provide an inter-thread control flow mechanism. For instance, if one thread should sleep until it receives some work to do, another thread can pass it some work, then signal it with a condition variable to keep it from having to constantly check for new input.

The API by example#

I’ll introduce the API via example. In the last chapter, there was an example that fetched a random comic URL from xkcd.com. Now you’ll extend that to do something with that URL.

require 'thread'
require 'net/http'

mutex    = Mutex.new
condvar  = ConditionVariable.new
results  = Array.new

Thread.new do
  10.times do
    response = Net::HTTP.get_response('dynamic.xkcd.com', '/random/comic/')
    random_comic_url = response['Location']

    mutex.synchronize do
      results << random_comic_url
      condvar.signal                     # Signal the ConditionVariable
    end
  end
end

comics_received = 0

until comics_received >= 10
  mutex.synchronize do
    while results.empty?
      condvar.wait(mutex)
    end

    url = results.shift
    puts "You should check out #{url}"
  end

  comics_received += 1
end

Let’s step through this bit by bit.

mutex    = Mutex.new
condvar  = ConditionVariable.new
results  = Array.new

Here you create a new ConditionVariable, in much the same way that you create a Mutex. This instance, too, must be shared amongst threads in order to be useful.

Thread.new do
  10.times do
    response = Net::HTTP.get_response('dynamic.xkcd.com', '/random/comic/')
    random_comic_url = response['Location']

    mutex.synchronize do
      results << random_comic_url
      condvar.signal                     # Signal the ConditionVariable
    end
  end
end

Here we have one thread fetching a random comic from xkcd, ten times. Since it’s just one thread, the HTTP requests are actually happening serially, one after another. This just puts that work in a background thread, so the main thread can continue on.

Once this thread receives a random comic URL, it locks the mutex, pushes the URL onto results, then signals condvar.

The ConditionVariable#signal method takes no parameters and has no meaningful return value. Its function is to wake up a thread that is waiting on itself. That happens lower down. If no threads are currently waiting, this is just a no-op.

comics_received = 0

until comics_received >= 10
  mutex.synchronize do
    while results.empty?
      condvar.wait(mutex)
    end

The background thread is fetching the random comics and, here, the main thread will be collecting the results.

First, some housekeeping to make it sure gets all 10 results.

The next part might seem a bit strange. The call to Mutex#synchronize will lock the mutex, then assuming that there are no results ready to be collected, ConditionVariable#wait is called. If results is not empty, it can be processed immediately without waiting.

At first glance, it seems that waiting on the condition variable while holding a mutex will block other threads from holding that mutex. But notice how you pass in the mutex to ConditionVariable#wait? It needs to receive a locked mutex.

ConditionVariable#wait will then unlock the Mutex and put the thread to sleep. When condvar is signaled, if this thread is first in line, it will wake up and lock the mutex again.

The last bit of strangeness here is the while loop. An if seems more natural here than a while. The reason for the while is a fail-safe. ConditionVariable#wait doesn’t receive any state information; it just notifies that an event happened.

If this thread wakes up, but some other thread has already removed the result it was expecting from results, it could go forward and find nothing waiting there. So once it reclaims the mutex, it rechecks its condition. If results is empty again, it can just go back to waiting for the next signal.

    url = results.shift
    puts "You should check out #{url}"
  end

  comics_received += 1
end

This part wraps things up. Still inside the loop, it shifts the oldest element from the Array and prints it.

Then notice that the counter is incremented using += outside the mutex. This is an instance where that operation doesn’t need protection. That variable is only visible to one thread; hence no synchronization is needed.

Broadcast#

There’s one other part of this small API we didn’t cover: broadcast.

There are two different methods that can signal threads:

1. ConditionVariable#signal will wake up exactly one thread that’s waiting on this ConditionVariable.
2. ConditionVariable#broadcast will wake up all threads currently waiting on this ConditionVariable.