7.6. Condition Variables¶

7.6.1. Condition Variables vs. Semaphores¶

Condition variables and semaphores appear very similar, as they both provide a mechanism that allow threads to signal that a custom event has occurred. But the differences between condition variables and semaphore signaling go beyond just a shift in terminology.

• The pthread_cond_wait() function performs multiple functions. It first releases the mutex and blocks until the corresponding signal is received; it then re-acquires the mutex that had been locked. Specifically, both of these actions are considered to occur atomically; unless an error occurs, the thread is guaranteed to have acquired the mutex by the time the function returns.
• Condition variables support broadcasting. The pthread_cond_broadcast() function will notify all threads that are waiting on the condition. Moreover, each thread will resume one at a time with the mutex acquired. With the additional mutual exclusion guarantees, condition variables can be combined in a thread-safe manner with other pieces of data to make the condition more meaningful.
• Condition variables are a standard part of the POSIX thread library, and they are more widely supported. For instance, some systems include the unnamed POSIX semaphore interface, but the implementation is empty, as named semaphores are preferred. There is no similar distinction in condition variables, and there is wider support for them.

7.6.2. How to Use a Condition Variable¶

There are several conventional practices for condition variables that may not be immediately obvious.

• A thread must acquire the mutex before calling pthread_cond_wait(), which will release the mutex. Calling pthread_cond_wait() without having locked the mutex leads to undefined behavior.
• Calls to pthread_cond_wait() should be made inside a while loop. The POSIX specification allows threads to wake up even if the signal was not sent (called a spurious wake up). By checking the return value of pthread_cond_wait(), the thread can determine if the wake up was spurious and go back to waiting if necessary.
• Just calling pthread_cond_signal() or pthread_cond_broadcast() is insufficient to wake up waiting threads, as the threads are locked by the mutex rather than the condition variable. The functions must be followed by a call to pthread_mutex_unlock(), which will allow pthread_cond_wait() to acquire the mutex and return.

7.6.3. A Condition Variable Example¶

The following sample program uses one thread read lines of keyboard input from STDIN, then passing on the information to two other threads. If the input is a string, the second thread gets the length of the string and adds it to a counter. If the input can be converted to an long using strtol(), then the integer value is added to the counter by the third thread.

The threads all rely on the following shared struct. The input_cond condition variable is used to indicate that a line of input has been received. The input_processed_cond variable is used to indicate that the two helper threads have processed the input and the keyboard listener can get more input. The other fields are used to pass information between the threads.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13

#define MAX_LENGTH 100

/* Common struct with pointers to all variables needed */
struct args {
  int counter;
  pthread_cond_t *input_cond;
  pthread_cond_t *input_processed_cond;
  pthread_mutex_t *mutex;
  char *buffer;
  long current_value;
  bool shutdown;
  bool input_is_string;
};

The keyboard listener starts by acquiring the mutex, guaranteeing that this thread has mutually exclusive access to the shared data. After reading a line of input with fgets(), this thread tries to convert the input to an long with strtol(). If so, it sets the current_value field to this integer value. If not, the string is checked against the string "shutdown", which is used to make the program stop. In all three cases, pthread_cond_broadcast() signals to the other threads that input has been received. The listener then waits on the input_processed_cond condition, which indicates that the input has been processed completely by another thread. Finally, if the shutdown message was received, this thread sets a boolean value that the others detect during another broadcast. Code Listing 7.15 shows the keyboard listener thread.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48

/* Code Listing 7.15:
   A thread that uses condition variable broadcasting to share data
 */

/* Keyboard listener thread; gets input and signals to helper 
   threads */
void *
keyboard_listener (void *_args)
{
  struct args *args = (struct args *) _args;

  /* With condvars, always acquire the lock before waiting */
  pthread_mutex_lock (args->mutex);
  while (1)
    {
      /* Get input from the user */
      printf ("Enter a string/number (or shutdown to quit): ");
      memset (args->buffer, 0, MAX_LENGTH + 1);
      if (fgets (args->buffer, MAX_LENGTH, stdin) == NULL)
        break;

      /* Check for a numeric value or "shutdown" */
      long guess = strtol (args->buffer, NULL, 10);
      if (guess != 0)
        {
          args->current_value = guess;
          args->input_is_string = false;
        }
      else if (strncmp (args->buffer, "shutdown", 8) != 0)
        args->input_is_string = true;
      else
        break;

      /* Send signal to all waiting threads */
      pthread_cond_broadcast (args->input_cond);

      /* Wait for confirmation the input is processed */
      pthread_cond_wait (args->input_processed_cond, args->mutex);
      flush_buffer (args->buffer);
    }

  /* Send the shutdown input condition */
  args->shutdown = true;
  pthread_cond_broadcast (args->input_cond);
  pthread_mutex_unlock (args->mutex);

  pthread_exit (NULL);
}

Code Listing 7.16 shows the additional threads that share this data. The count_chars() and add_number() threads behave in approximately the same manner. They both start by acquiring the mutex, then waiting on the input_cond. In both cases, the call to pthread_cond_wait() releases the lock at this point. Once the signal has been received and the mutex is re-acquired, the threads check if they need to shutdown. If not, they each check if the input was a string. Only the count_chars() thread processes string input, whereas only the add_number() thread processes numeric input. If the thread is not supposed to process input, it uses continue to go back to the beginning of the loop and wait on the condition again. If the thread did process its appropriate input, it signals on the input_processed_cond variable, which allows keyboard_listener() to move on to reading the next line of input.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39
40
41
42
43
44
45
46
47
48
49
50
51
52
53
54
55
56
57
58
59
60
61
62
63
64
65
66
67
68
69
70
71
72

/* Code Listing 7.16:
   Threads that process data as it is received via a condition variable
 */

/* Thread for counting characters in a string */
void *
count_chars (void *_args)
{
  struct args *args = (struct args *) _args; 
  /* With condvars, always acquire the lock before waiting */
  pthread_mutex_lock (args->mutex);
  while (1)
    {
      /* Wait on the input condition */
      pthread_cond_wait (args->input_cond, args->mutex);
      /* Signal received, mutex has been re-acquired */

      /* If the input was shutdown signal, quit */
      if (args->shutdown) break;

      /* If the input was a number, ignore this signal */
      if (!args->input_is_string) continue;

      /* Input was string not shutdown; get length without \n */
      char *newline = strchr (args->buffer, '\n');
      if (newline != NULL) *newline = '\0';
      args->counter += strlen (args->buffer);

      /* Restore the newline for buffer flushing */
      if (newline != NULL) *newline = '\n';

      /* Signal back to keyboard listener that input processing 
         is done */
      pthread_cond_signal (args->input_processed_cond); 
      /* Do NOT unlock mutex; pthread_cond_wait() will do that */
    }

  /* Shutting down. Send acknowledgement signal back */
  pthread_cond_signal (args->input_processed_cond);
  pthread_mutex_unlock (args->mutex);
  pthread_exit (NULL);
}

void *
add_number (void *_args)
{
  struct args *args = (struct args *) _args; 
  /* With condvars, always acquire the lock before waiting */
  pthread_mutex_lock (args->mutex);
  while (1)
    {
      /* Wait on the input condition */
      pthread_cond_wait (args->input_cond, args->mutex);
      /* Signal received, mutex has been re-acquired */

      /* If the input was shutdown signal, quit */
      if (args->shutdown) break;

      /* If the input was a string, ignore this signal */
      if (args->input_is_string) continue;

      /* Add current value to the counter and send signal back */
      args->counter += args->current_value;
      pthread_cond_signal (args->input_processed_cond); 
      /* Do NOT unlock mutex; pthread_cond_wait() will do that */
    }

  /* Shutting down. Send acknowledgement signal back */
  pthread_cond_signal (args->input_processed_cond);
  pthread_mutex_unlock (args->mutex);
  pthread_exit (NULL);
}

Bug Warning

---

For completeness, both the count_chars() and add_number() threads should check the return value of pthread_cond_wait() to determine if the signal was spurious. However, this code omits this check for brevity and algorithmic clarity.

Finally, Code Listing 7.17 illustrates how to initialize the condition variables, create the threads, and clean up all of the resources for the condition variables and the mutex.

 1
 2
 3
 4
 5
 6
 7
 8
 9
10
11
12
13
14
15
16
17
18
19
20
21
22
23
24
25
26
27
28
29
30
31
32
33
34
35
36
37
38
39

/* Code Listing 7.17:
   Initializing a condition variable and other data structures
 */

pthread_t threads[3];
pthread_cond_t input_cond;
pthread_cond_t input_processed_cond;
pthread_mutex_t mutex;
char buffer[MAX_LENGTH + 1];

/* Initialize the mutex and the condition variables */
pthread_mutex_init (&mutex, NULL);
pthread_cond_init (&input_cond, NULL);
pthread_cond_init (&input_processed_cond, NULL);

/* Set up the args for all threads */
struct args args;
args.counter = 0;
args.input_cond = &input_cond;
args.input_processed_cond = &input_processed_cond;
args.mutex = &mutex;
args.buffer = buffer;
args.current_value = 0;
args.shutdown = false;
args.input_is_string = false;

/* Create and join the threads */
assert (pthread_create (&threads[0], NULL, count_chars, &args) == 0);
assert (pthread_create (&threads[1], NULL, add_number, &args) == 0);
assert (pthread_create (&threads[2], NULL, keyboard_listener, &args) == 0);
pthread_join (threads[0], NULL);
pthread_join (threads[1], NULL);
pthread_join (threads[2], NULL);

/* Print out total, destroy the mutex and condition variables */
printf ("Total: %d\n", args.counter);
pthread_mutex_destroy (&mutex);
pthread_cond_destroy (&input_cond);
pthread_cond_destroy (&input_processed_cond);

7.6.4. Monitors and Synchronized Methods¶

Figure 7.6.3: Architecture of a monitor with synchronized data access

The preceding code is an example of an object-oriented construct known as a monitor. Figure 7.6.3 illustrates the general architecture of a monitor. Specifically, a monitor is a class or data structure that combines condition variables, mutexes, and other application-specific data. All of the internal data is considered private to the monitor, and other pieces of code can only interact with the monitor by invoking a method on the object.

The key characteristic of a monitor is that all methods are mutually exclusive in execution. That is, as with the thread functions above, methods in a monitor begin by locking the monitor’s mutex. Doing so guarantees that only one thread is in the monitor at any given moment. The method ends by releasing the mutex, allowing other threads to enter the monitor.

Within the monitor, there are associated condition variables to synchronize access between the monitor and the general execution environment. While the methods are executing, they can use these condition variables to detect key events or to check for safe conditions. If the condition variable check (i.e., pthread_cond_wait()) fails, then the thread must release the mutex and exit the monitor. The thread then waits in the condition variable’s associated waiting queue until it can resume execution.

Readers familiar with the Java synchronized keyword have been essentially using monitors by a different name. That is, the primary function of the synchronized keyword is to have injected code that acquires a hidden mutex at the beginning of the method execution, then releases it when returning.