Synchronization II: semaphores, condition variables, monitors and classic problems

Locks give mutual exclusion: only one thread at a time in a critical section. Many concurrency problems need more than that. A consumer thread must wait until a producer has put something in a queue; a pool of database connections must admit at most ten users at once; all the threads in a simulation must finish step 41 before any starts step 42. These are problems of ordering and waiting, not just exclusion. This lesson covers the tools operating systems and libraries provide for them: semaphores, condition variables, monitors and barriers, and applies them to the classic problems every systems programmer should recognize.

Lesson 5 covered races, locks and atomic operations; Lesson 7 covers deadlock, the main hazard of combining several locks. CMPUT-274 met Python's Queue, Event and Lock objects; here we use the POSIX primitives that underlie such libraries.

Semaphores

Edsger Dijkstra introduced the semaphore in the 1960s. A semaphore is an integer counter with two atomic operations. P, also called wait or down (from the Dutch 'proberen', to try), decrements the counter if it is positive and otherwise blocks until it can. V, also called signal, post or up ('verhogen', to raise), increments the counter and wakes one waiting thread if there is one. A binary semaphore, initialized to 1, behaves like a lock. A counting semaphore initialized to n admits up to n threads at once, which models a pool of n identical resources: database connections, licence seats or download slots.

#include <pthread.h>
#include <semaphore.h>
#include <stdio.h>
#include <unistd.h>

#define WORKERS 10
static sem_t slots;                               /* counting semaphore: 3 permits */
static pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
static int inside = 0, peak = 0;

static void *download(void *arg) {
    (void)arg;
    sem_wait(&slots);                             /* P: take a permit or sleep */
    pthread_mutex_lock(&m);
    if (++inside > peak) peak = inside;
    pthread_mutex_unlock(&m);
    usleep(20000);                                /* the limited resource is in use */
    pthread_mutex_lock(&m); inside--; pthread_mutex_unlock(&m);
    sem_post(&slots);                             /* V: return the permit */
    return NULL;
}

int main(void) {
    sem_init(&slots, 0, 3);
    pthread_t t[WORKERS];
    for (int i = 0; i < WORKERS; i++) pthread_create(&t[i], NULL, download, NULL);
    for (int i = 0; i < WORKERS; i++) pthread_join(t[i], NULL);
    printf("%d downloads ran; at most %d ran at the same time (limit 3)\n", WORKERS, peak);
    sem_destroy(&slots);
    return 0;
}

$ gcc -O2 -Wall -g -pthread -fsanitize=address,undefined toy.c -o toy
$ ./toy
10 downloads ran; at most 3 ran at the same time (limit 3)

Semaphores are also useful for signalling between threads. A semaphore initialized to 0 lets one thread wait for an event that another thread will announce: the waiter calls P and sleeps; the announcer calls V. Unlike a lock, a semaphore has no owner, so the thread that calls V need not be the one that called P. This flexibility is also its weakness. Semaphores are easy to misuse: a forgotten V leaks a permit forever, an extra V adds a permit that should not exist, and nothing in the code makes the intended pattern obvious. POSIX provides unnamed semaphores (sem_init) for threads and named semaphores (sem_open) for separate processes.

Condition variables

A condition variable lets a thread wait, while holding a lock, until some condition on shared data becomes true. It always works together with a mutex. The waiting thread locks the mutex, checks the condition and, if it is false, calls pthread_cond_wait, which atomically releases the mutex and puts the thread to sleep; when woken, the thread reacquires the mutex before returning. A thread that changes the shared data calls pthread_cond_signal to wake one waiter or pthread_cond_broadcast to wake all of them.

Two rules make condition variables work. First, always test the condition in a while loop, never an if: a woken thread must re-check because another thread may have run first and changed the state, and POSIX allows spurious wake-ups with no signal at all. Second, change the shared state and signal while holding the mutex (or at least change it under the mutex), so that no waiter can test the condition between the change and the signal and miss the wake-up. Condition variables are memoryless: a signal sent when nobody is waiting is lost, which is why the condition itself, not the signal, carries the information.

The bounded-buffer (producer-consumer) problem

In the bounded-buffer problem, producer threads put items into a buffer of fixed size and consumer threads take them out. Producers must wait when the buffer is full and consumers when it is empty, and the buffer's indexes must be updated under mutual exclusion. The pattern is everywhere: a pipe between two processes, a print spooler, a web server's queue of incoming connections, a logging thread draining messages from application threads, and the kernel's queue of disk requests.

#include <pthread.h>
#include <stdio.h>

#define CAP 4                     /* bounded buffer of 4 slots */
#define ITEMS 10
static int buf[CAP], head = 0, tail = 0, count = 0;
static pthread_mutex_t m = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t not_full = PTHREAD_COND_INITIALIZER;
static pthread_cond_t not_empty = PTHREAD_COND_INITIALIZER;
static int max_seen = 0;

static void *producer(void *arg) {
    (void)arg;
    for (int i = 1; i <= ITEMS; i++) {
        pthread_mutex_lock(&m);
        while (count == CAP)                      /* always re-check in a loop */
            pthread_cond_wait(&not_full, &m);     /* releases m while sleeping */
        buf[tail] = i * i; tail = (tail + 1) % CAP; count++;
        if (count > max_seen) max_seen = count;
        pthread_cond_signal(&not_empty);
        pthread_mutex_unlock(&m);
    }
    return NULL;
}

static void *consumer(void *arg) {
    (void)arg;
    for (int i = 1; i <= ITEMS; i++) {
        pthread_mutex_lock(&m);
        while (count == 0)
            pthread_cond_wait(&not_empty, &m);
        int v = buf[head]; head = (head + 1) % CAP; count--;
        pthread_cond_signal(&not_full);
        pthread_mutex_unlock(&m);
        printf("consumed %3d\n", v);              /* print outside the lock */
    }
    return NULL;
}

int main(void) {
    pthread_t p, c;
    pthread_create(&c, NULL, consumer, NULL);
    pthread_create(&p, NULL, producer, NULL);
    pthread_join(p, NULL); pthread_join(c, NULL);
    printf("buffer never held more than %d of %d slots\n", max_seen, CAP);
    return 0;
}

$ gcc -O2 -Wall -g -pthread -fsanitize=thread toy.c -o toy
$ ./toy
consumed   1
consumed   4
consumed   9
consumed  16
consumed  25
consumed  36
consumed  49
consumed  64
consumed  81
consumed 100
buffer never held more than 4 of 4 slots

The output shows every item consumed in order and the buffer never overfilling. Because the consumer prints outside the lock, slow output does not hold up the producer. The same problem can be solved with three semaphores: one counting empty slots, initialized to the capacity; one counting full slots, initialized to 0; and a binary semaphore for mutual exclusion. Getting the order of operations right matters there: if a producer takes the mutex before waiting for an empty slot, it can sleep while holding the mutex and block the consumer forever, a deadlock.

Monitors

A monitor packages shared data, the operations on it and the synchronization into one unit, so that the language or library guarantees that only one thread is active inside the monitor at a time and provides condition variables for waiting. Tony Hoare and Per Brinch Hansen proposed monitors in the 1970s to make concurrent programs easier to get right than raw semaphores. Java builds monitors into every object: synchronized methods lock the object, and wait, notify and notifyAll are its condition variable. C# has lock and Monitor, Python has threading.Condition, and Go's channels offer a different style based on passing messages rather than sharing memory.

A subtle design choice is what happens when a thread signals. In Hoare semantics, the woken thread runs immediately and the signaller waits, so the condition is guaranteed true when the waiter resumes. In Mesa semantics, used by Java, POSIX and nearly every real system, the signaller keeps running and the woken thread merely becomes ready, so by the time it runs the condition may be false again. That is the deeper reason for the while loop around every wait.

Readers and writers

Many shared data structures are read far more often than written: a routing table, a configuration object, a cache. A plain mutex makes readers wait for each other unnecessarily. A reader-writer lock allows any number of readers at once, or one writer alone. The design question is fairness. With reader preference, new readers may enter while a writer waits, so a steady stream of readers can starve writers. With writer preference, a waiting writer blocks new readers, which bounds the writer's wait but can delay readers. POSIX provides pthread_rwlock_t, and Linux's kernel adds seqlocks and RCU, which let readers proceed without any locking at all.

# Readers-writers on a timeline: (name, arrival ms, duration ms). Compare the two fairness policies.
arrivals = [("R1", 0, 4), ("R2", 1, 4), ("W1", 2, 3), ("R3", 3, 4), ("R4", 4, 4), ("R5", 6, 4), ("R6", 8, 4)]
def simulate(prefer_writers):
    waiting, active, start = [], {}, {}
    for t in range(60):
        for n, a, d in arrivals:
            if a == t: waiting.append((n, d))
        for n in [n for n, end in active.items() if end == t]: del active[n]
        writer_active = any(n[0] == "W" for n in active)
        writer_waiting = any(n[0] == "W" for n, _ in waiting)
        for n, d in list(waiting):
            if n[0] == "W" and not active:
                active[n] = t + d; start[n] = t; waiting.remove((n, d)); break
            if n[0] == "R" and not writer_active and not (prefer_writers and writer_waiting):
                active[n] = t + d; start[n] = t; waiting.remove((n, d))
    return start
for pref in (False, True):
    start = simulate(pref)
    order = " ".join(f"{n}@{start[n]}" for n in sorted(start, key=lambda n: (start[n], n)))
    w_wait = start["W1"] - 2
    print(f"{'writer' if pref else 'reader'} preference: {order}")
    print(f"   W1 waited {w_wait} ms")

reader preference: R1@0 R2@1 R3@3 R4@4 R5@6 R6@8 W1@12
   W1 waited 10 ms
writer preference: R1@0 R2@1 W1@5 R3@8 R4@8 R5@8 R6@8
   W1 waited 3 ms

Barriers

A barrier makes a group of threads wait until all of them reach the same point. It is the natural tool for computations that proceed in phases, such as a physics simulation in which every thread must finish updating its part of the grid before any thread reads its neighbours' values for the next step. POSIX provides pthread_barrier_t; OpenMP and MPI have their own. The cost of a barrier is that every thread waits for the slowest, so balancing work between threads matters.

#define _POSIX_C_SOURCE 200809L   /* barriers are an optional POSIX feature */
#include <pthread.h>
#include <stdio.h>

#define T 3
static pthread_barrier_t bar;
static int phase_done[T][2];

static void *worker(void *arg) {
    int id = *(int *)arg;
    phase_done[id][0] = 1;                    /* phase 1 work */
    pthread_barrier_wait(&bar);               /* nobody starts phase 2 until all finish phase 1 */
    int all = 1;
    for (int k = 0; k < T; k++) all &= phase_done[k][0];
    phase_done[id][1] = all;                  /* phase 2 can rely on every phase 1 result */
    return NULL;
}

int main(void) {
    pthread_t t[T]; int ids[T];
    pthread_barrier_init(&bar, NULL, T);
    for (int i = 0; i < T; i++) { ids[i] = i; pthread_create(&t[i], NULL, worker, &ids[i]); }
    for (int i = 0; i < T; i++) pthread_join(t[i], NULL);
    for (int i = 0; i < T; i++)
        printf("worker %d saw every phase-1 result before phase 2: %s\n", i, phase_done[i][1] ? "yes" : "no");
    pthread_barrier_destroy(&bar);
    return 0;
}

$ gcc -O2 -Wall -g -pthread -fsanitize=thread toy.c -o toy
$ ./toy
worker 0 saw every phase-1 result before phase 2: yes
worker 1 saw every phase-1 result before phase 2: yes
worker 2 saw every phase-1 result before phase 2: yes

Message passing as an alternative

All of the tools so far coordinate threads that share memory. Another school of thought says: do not communicate by sharing memory; share memory by communicating. In the message-passing style, threads or processes own their data privately and send each other messages through channels or queues. Go's channels, Erlang's mailboxes, the actor model in Akka, and Unix pipes from Lesson 2 all follow this idea. The bounded buffer reappears inside the channel implementation, but application code no longer touches locks directly, which removes whole classes of bugs. Message passing also extends naturally across machines, the subject of CMPUT-450.

Worked examples