The key issue when tasks are cooperating is handshaking between those tasks. To accomplish this handshaking, we use the same foundation: the mutex, which in this case guarantees that only one task can respond to a signal. This eliminates any possible race conditions. On top of the mutex, we add a way for a task to suspend itself until some external state changes ("the plumbing is now in place"), indicating that it’s time for that task to move forward. In this section, we’ll look at the issues of handshaking between tasks, the problems that can arise, and their solutions.
Wait and signal
In ZThreads, the basic class that uses a mutex and allows task suspension is the Condition, and you can suspend a task by calling wait( ) on a Condition. When external state changes take place that might mean that a task should continue processing, you notify the task by calling signal( ), to wake up one task, or broadcast( ), to wake up all tasks that have suspended themselves on that Condition object.
There are two forms of wait( ). The first takes an argument in milliseconds that has the same meaning as in sleep( ): "pause for this period of time." The difference is that in a timed wait( ):
1. The Mutex that is controlled by the Condition object is released during the wait( ).
2. You can come out of the wait( ) due to a signal( ) or a broadcast( ), as well as by letting the clock run out.
The second form of wait( ) takes no arguments; this version is more commonly used. It also releases the mutex, but this wait( ) suspends the thread indefinitely until that Condition object receives a signal( ) or broadcast( ).
Typically, you use wait( ) when you’re waiting for some condition to change that is under the control of forces outside the current function. (Often, this condition will be changed by another thread.) You don’t want to idly loop while testing the condition inside your thread; this is called a "busy wait," and it’s a bad use of CPU cycles. Thus, wait( ) allows you to suspend the thread while waiting for the world to change, and only when a signal( ) or broadcast( ) occurs (suggesting that something of interest may have happened), does the thread wake up and check for changes. Therefore, wait( ) provides a way to synchronize activities between threads.
Let’s look at a simple example. WaxOMatic.cpp applies wax to a Car and polishes it using two separate processes. The polishing process cannot do its job until the application process is finished, and the application process must wait until the polishing process is finished before it can put on another coat of wax. Both WaxOn and WaxOff use the Car object, which contains a Condition that it uses to suspend a thread inside waitForWaxing( ) or waitForBuffing( ):
//: C11:WaxOMatic.cpp
// Basic thread cooperation.
//{L} ZThread
#include "zthread/Thread.h"
#include "zthread/Mutex.h"
#include "zthread/Guard.h"
#include "zthread/Condition.h"
#include "zthread/ThreadedExecutor.h"
#include <iostream>
#include <string>
using namespace ZThread;
using namespace std;
class Car {
Mutex lock;
Condition condition;
bool waxOn;
public:
Car() : condition(lock), waxOn(false) {}
void waxed() {
Guard<Mutex> g(lock);
waxOn = true; // Ready to buff
condition.signal();
}
void buffed() {
Guard<Mutex> g(lock);
waxOn = false; // Ready for another coat of wax
condition.signal();
}
void waitForWaxing() {
Guard<Mutex> g(lock);
while(waxOn == false)
condition.wait();
}
void waitForBuffing() {
Guard<Mutex> g(lock);
while(waxOn == true)
condition.wait();
}
};
class WaxOn : public Runnable {
CountedPtr<Car> car;
public:
WaxOn(CountedPtr<Car>& c) : car(c) {}
void run() {
try {
while(!Thread::interrupted()) {
cout << "Wax On!" << endl;
Thread::sleep(200);
car->waxed();
car->waitForBuffing();
}
} catch(Interrupted_Exception&) { /* Exit */ }
cout << "Ending Wax On process" << endl;
}
};
class WaxOff : public Runnable {
CountedPtr<Car> car;
public:
WaxOff(CountedPtr<Car>& c) : car(c) {}
void run() {
try {
while(!Thread::interrupted()) {
car->waitForWaxing();
cout << "Wax Off!" << endl;
Thread::sleep(200);
car->buffed();
}
} catch(Interrupted_Exception&) { /* Exit */ }
cout << "Ending Wax Off process" << endl;
}
};
int main() {
cout << "Press <Enter> to quit" << endl;
try {
CountedPtr<Car> car(new Car);
ThreadedExecutor executor;
executor.execute(new WaxOff(car));
executor.execute(new WaxOn(car));
cin.get();
executor.interrupt();
} catch(Synchronization_Exception& e) {
cerr << e.what() << endl;
}
} ///:~
In Car’s constructor, a single Mutex is wrapped in a Condition object so that it can be used to manage inter-task communication. However, the Condition object contains no information about the state of your process, so you need to manage additional information to indicate process state. Here, Car has a single bool waxOn, which indicates the state of the waxing-polishing process.
In waitForWaxing( ), the waxOn flag is checked, and if it is false, the calling thread is suspended by calling wait( ) on the Condition object. It’s important that this occur inside a guarded clause, in which the thread has acquired the lock (here, by creating a Guard object). When you call wait( ), the thread is suspended and the lock is released. It is essential that the lock be released because, to safely change the state of the object (for example, to change waxOn to true, which must happen if the suspended thread is to ever continue), that lock must be available to be acquired by some other task. In this example, when another thread calls waxed( ) to tell it that it’s time to do something, the mutex must be acquired in order to change waxOn to true. Afterward, waxed( ) sends a signal( ) to the Condition object, which wakes up the thread suspended in the call to wait( ). Although signal( ) may be called inside a guarded clause—as it is here—you are not required to do this.[127]
127
This is in contrast to Java, where you must hold the lock in order to call notify() (Java’s version of signal()). Although Posix threads, on which the ZThread library is loosely based, do not require that you hold the lock in order to call signal() or broadcast(), it is often recommended.