139. POSIX threads and synchronization: mutex, cond, rwlock, barrier, TLS

Processes (section 136) are isolated from each other, and exchanging data between them costs a copy through the kernel. Threads live inside one process: shared memory, shared files, a shared heap — and each has its own stack, its own registers and its own signal mask. Creating a thread is cheaper than creating a process, and there is no need to pass data to it — the data is already shared. The price is races: two threads changing one variable corrupt it, and you have to protect shared data yourself. This section covers POSIX threads (pthread) and the full set of synchronization tools, each in an experiment from _verify/sys139/ in stm-sys:1.

FreeRTOS tasks from part IV are the same threads: the mutex, semaphore and queue there work the same way, only the names differ, and everything said here about races and deadlocks applies to firmware too.

Terms used in the documentation

Term	Meaning
thread	a flow of execution inside a process
joinable, detached	a thread someone waits for with join; a thread nobody waits for
mutex (mutual exclusion)	a lock that lets one thread in at a time
critical section	code that only one thread may run at a time
condition variable	a way to sleep until an event happens
spurious wakeup	a wakeup without a signal
read-write lock	a lock for "many readers, one writer"
spin lock	a lock that waits by busy-looping
barrier	a meeting point for threads
thread-local storage (TLS)	data of which each thread has its own copy
deadlock	threads waiting for each other forever
cancellation point	a place where a thread can be cancelled
data race	unsynchronized access to shared data
Amdahl's law	the limit of parallel speedup

139.1 Create, wait for, let go

    for (int k = 0; k < 4; k++) pthread_create(&th[k], NULL, fn, NULL);
    for (int k = 0; k < 4; k++) pthread_join(th[k], NULL);

pthread_create takes a function void *f(void *) and an argument; the thread returns a pointer, which pthread_join receives. Returning the address of a thread's local variable is not allowed: the thread's stack disappears together with it. pthread functions do not write to errno; they return an error code (each thread has its own errno).

There are three ways to end a thread: return from the function, pthread_exit, and pthread_cancel from another thread — the last one takes effect only at cancellation points (sleep, read, pthread_cond_wait…). A finished thread that nobody has joined holds resources, like a zombie; pthread_detach says that nobody will wait for it, and its resources are freed automatically. exit from any thread terminates the whole process.

fork in a multithreaded program. fork copies only the thread that called it. A mutex that another thread was holding at that moment stays locked in the child: there is no one left to release it. Experiment: a thread took a mutex, the main thread called fork — in the child pthread_mutex_timedlock waited a second and returned ETIMEDOUT. glibc restores its own malloc and stdio locks in the child by itself — 300 children with a neighbouring thread spinning on malloc never hung — but it knows nothing about other libraries' locks. So in the child of a multithreaded program, before exec, call only async-signal-safe functions (136.6), or better, call exec or posix_spawn right away; for special cases there is pthread_atfork.

Cost: 1000 create + join pairs take 46 ms, 46 µs each; a fork + wait pair cost 116 µs (section 136). The default thread stack is the size of RLIMIT_STACK, 8 MB; a thousand threads mean 8 GB of address space. Physically the stack occupies only the pages it has touched, and on a 64-bit machine this is usually not a problem, but on a 32-bit Linux board, where a program has about 3 GB available, only a few hundred such threads fit. So the size is set through attributes (pthread_attr_setstacksize).

A signal sent to a process is delivered to any thread that does not block it. So signals are blocked in main with pthread_sigmask before the threads are created (threads inherit the mask) and received in one thread with sigwait or signalfd (141.4).

139.2 A race and three ways to close it

Four threads each increment a shared counter a million times:

static void *f_plain(void *a)   { (void)a; for (int i = 0; i < N; i++) plain++; return NULL; }
static void *f_mutex(void *a)   { (void)a; for (int i = 0; i < N; i++) { pthread_mutex_lock(&mx); guarded++; pthread_mutex_unlock(&mx); } return NULL; }
static void *f_atomic(void *a)  { (void)a; for (int i = 0; i < N; i++) atomic_fetch_add(&atomic_cnt, 1); return NULL; }

4 потока × 1000000 приращений:
  без защиты: 1455881 (потеряно 2544119), 2 мс
  мьютекс:    4000000, 178 мс
  atomic:     4000000, 35 мс

Without protection most of the increments are lost (in this run, more than half): plain++ is a read, an addition and a write, and two threads that read the same value write the same result. The variable is declared volatile — and that changed nothing: volatile forbids the compiler to remove memory accesses, but it does not provide atomicity (section 124). Without volatile the compiler at -O2 would fold the million increments into one addition, and the race would be invisible — that is the whole difference.

The mutex closes the race at a cost of 178 ms — tens of times more expensive: every contended acquire is a trip into the kernel (futex). The atomic addition takes 35 ms: one processor instruction with the lock prefix: the core takes ownership of the cache line, and under contention moving that line between cores is what costs those 35 ms. The rule of choice: a counter or a flag — an atomic variable; a structure, a list, several fields at once — a mutex.

Races are found by tools, not by eye. ThreadSanitizer (-fsanitize=thread) names the line:

WARNING: ThreadSanitizer: data race (pid=14)
SUMMARY: ThreadSanitizer: data race /v/sys139/race_mutex.c:15 in f_plain

Helgrind from Valgrind (valgrind --tool=helgrind) does the same without rebuilding, but tens of times slower:

Possible data race during read of size 8 at 0x10C060 by thread #3
ERROR SUMMARY: 6 errors from 2 contexts

139.3 Mutex: types and deadlock

pthread_mutex_t is 40 bytes; PTHREAD_MUTEX_INITIALIZER is for static ones, pthread_mutex_init with attributes for the rest. The type is set by an attribute:

The classic deadlock: thread 1 holds a and waits for b, thread 2 holds b and waits for a. With a plain lock both wait forever; the program uses pthread_mutex_timedlock and gets an answer after half a second:

    pthread_mutex_lock(&a); usleep(50000);                 /* держим a, тянемся к b */
    int r = lock_with_timeout(&b, 500);
    printf("поток 1: a взят, b: %s\n", r ? strerror(r) : "взят");

поток 1: a взят, b: Connection timed out
поток 2: b взят, a: Connection timed out
ERRORCHECK, второй lock из того же потока: Resource deadlock avoided
ERRORCHECK, лишний unlock: Operation not permitted

The cure for a deadlock is not a timeout but an order: all threads take the mutexes in the same order (a, then b), and the cycle never closes. ERRORCHECK is worth enabling in a debug build: it turns a hang into an error code.

139.4 Condition variable: a job queue

A mutex protects data but cannot wait for the event "data has arrived". That is what a condition variable paired with a mutex is for. A producer puts jobs into a list, and three consumers take them:

static void *consumer(void *arg)
{
    int me = (int)(long)arg;
    for (;;) {
        pthread_mutex_lock(&qlock);
        while (!head && !done)                       /* while, не if: пробуждение не гарантирует данных */
            pthread_cond_wait(&qready, &qlock);      /* отпускает мьютекс на время ожидания */
        if (!head && done) { pthread_mutex_unlock(&qlock); return NULL; }
        struct job *j = head; head = j->next;
        pthread_mutex_unlock(&qlock);
        processed[me]++;                             /* работа — вне мьютекса */
        free(j);
    }
}

обработано: 2346 + 1 + 653 = 3000

Three rules without which the queue breaks:

1. Wait in a while loop. pthread_cond_wait may return without a signal (a spurious wakeup), and another consumer may have taken the job first — the condition is checked again.
2. cond_wait releases the mutex while sleeping and takes it back before returning — so it is called only with the mutex held.
3. signal wakes one, broadcast wakes all. When closing the queue, set the done flag under the mutex and call broadcast, otherwise two of the three consumers sleep forever.

The 2346/1/653 split is not an error: threads are not obliged to share the work equally; whoever woke up first took the job. All 3000 were processed, none was lost and none was processed twice.

139.5 rwlock, spin lock, barrier

An rwlock lets many readers in at once and one writer alone. Four readers with 200 000 reads each and a writer with 1000 writes over the whole table — the table is uniform after every write:

rwlock: 4 читателя × 200000, писатель 1000 записей; таблица цела: да

It pays off when there are many reads and few writes; with frequent writes it loses to a mutex — readers pile up in the queue before each write.

A spin lock (pthread_spin_lock) does not sleep but spins in a loop until the lock is free. It is justified only on a multicore machine for sections lasting tens of nanoseconds; on a single core it is harmful: the thread spins until the owner gets a time slice.

A barrier gathers count threads: pthread_barrier_wait does not return until all of them have arrived. In the experiment five threads (four readers and the writer) start from one point, so that the measurement does not depend on the order of creation.

pthread_cond_t is 48 bytes, pthread_rwlock_t is 56.

139.6 Each thread's own data

A global variable is shared; a local one exists only inside a function. Between them is thread data: each thread has its own copy of one variable.

static __thread int tls_counter;                     /* у каждого потока своя копия */

Each of the four readers counted 200 000 in it, while in the main thread it stayed 0:

читатель: tls_counter = 200000, key = 0
главный поток: tls_counter = 0 (своя копия)

__thread (in C11, _Thread_local, section 140) is a static declaration. The dynamic variant is keys: pthread_key_create with a destructor function, pthread_setspecific/pthread_getspecific; the key is created once per process through pthread_once. Libraries that need their own copy of data in each thread do it this way. errno in glibc is built on __thread, while strtok_r is safe in a different way: its state is kept by the caller in the caller's own variable (section 140.5).

Cancellation and cleanup. pthread_cancel stops a thread at the nearest cancellation point, and before it dies the handlers registered with pthread_cleanup_push run — in reverse order. They are called on cancellation and on pthread_exit, and in the normal flow they are removed by the matching pthread_cleanup_pop (with argument 1, the handler is called). push and pop are placed as a pair in one scope; you must not leave through return between them.

cleanup: освободить ресурсы отменённого потока
отменённый поток вернул PTHREAD_CANCELED: да

139.7 How much parallelism gives: Amdahl's law

If a fraction a of the work cannot be parallelized, p threads speed the program up by no more than 1 / (a + (1 − a) / p) times. A sum of 40 million square roots on a 24-core machine:

ядер: 24; 1 поток: 46 мс
2 потоков: 23 мс, ускорение 1.96, Амдал при a=0,05: 1.90
4 потоков: 11 мс, ускорение 4.00, Амдал при a=0,05: 3.48
8 потоков: 6 мс, ускорение 7.12, Амдал при a=0,05: 5.93

The task is almost perfectly parallel — the threads share nothing except the result at the end — and the speedup is close to the number of threads. Add a shared mutex to the loop, and the sequential fraction grows, and eight threads give less than two. At a = 0.05 the limit is 20 times, no matter how many cores there are.

139.8 Pitfalls

139.9 Summary

• A thread: 46 µs to create versus 116 for a process; shared memory, its own 8 MB stack.
• fork in a multithreaded program leaves one thread, and other threads' mutexes stay locked: call exec or posix_spawn right away.
• A race loses increments (in the run, more than half); a mutex costs 178 ms, an atomic operation 35 ms; volatile is not protection.
• Take mutexes in one order; ERRORCHECK and timedlock for debugging.
• Condition variable: while, with the mutex, broadcast on closing.
• rwlock — many readers; barrier — a common start; __thread — a copy per thread.
• Amdahl's law: with a 5 % sequential part the limit is 20 times.

Full sources

race_mutex.c — a counter from four threads without protection, with a mutex and with an atomic addition.

/* Раздел 139: счётчик из четырёх потоков — без мьютекса, с мьютексом, с атомарным сложением;
   время каждого варианта. */
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <stdatomic.h>
#include <time.h>

#define N 1000000
static volatile long plain;                  /* volatile: чтобы -O2 не свернул цикл в одно сложение */
static long guarded;
static atomic_long atomic_cnt;
static pthread_mutex_t mx = PTHREAD_MUTEX_INITIALIZER;

static void *f_plain(void *a)   { (void)a; for (int i = 0; i < N; i++) plain++; return NULL; }
static void *f_mutex(void *a)   { (void)a; for (int i = 0; i < N; i++) { pthread_mutex_lock(&mx); guarded++; pthread_mutex_unlock(&mx); } return NULL; }
static void *f_atomic(void *a)  { (void)a; for (int i = 0; i < N; i++) atomic_fetch_add(&atomic_cnt, 1); return NULL; }

static double run4(void *(*fn)(void *))
{
    struct timespec t0, t1;
    pthread_t th[4];
    clock_gettime(CLOCK_MONOTONIC, &t0);
    for (int k = 0; k < 4; k++) pthread_create(&th[k], NULL, fn, NULL);
    for (int k = 0; k < 4; k++) pthread_join(th[k], NULL);
    clock_gettime(CLOCK_MONOTONIC, &t1);
    return (t1.tv_sec - t0.tv_sec) * 1e3 + (t1.tv_nsec - t0.tv_nsec) / 1e6;
}

int main(void)
{
    double a = run4(f_plain), b = run4(f_mutex), c = run4(f_atomic);
    printf("4 потока × %d приращений:\n", N);
    printf("  без защиты: %ld (потеряно %ld), %.0f мс\n", plain, 4L * N - plain, a);
    printf("  мьютекс:    %ld, %.0f мс\n", guarded, b);
    printf("  atomic:     %ld, %.0f мс\n", atomic_load(&atomic_cnt), c);
    return 0;
}

deadlock.c — a deadlock of two threads and its detection.

/* Раздел 139: взаимоблокировка двух потоков на двух мьютексах и её обнаружение:
   pthread_mutex_timedlock вместо вечного ожидания; ERRORCHECK ловит повторный захват. */
#include <stdio.h>
#include <string.h>
#include <pthread.h>
#include <time.h>
#include <errno.h>
#include <unistd.h>

static pthread_mutex_t a = PTHREAD_MUTEX_INITIALIZER, b = PTHREAD_MUTEX_INITIALIZER;
static pthread_barrier_t bar;   /* оба потока держат свой первый мьютекс, прежде чем тянуться ко второму */
static pthread_barrier_t done;  /* и отпускают его, только когда оба получили ответ timedlock */

static int lock_with_timeout(pthread_mutex_t *m, int ms)
{
    struct timespec ts; clock_gettime(CLOCK_REALTIME, &ts);
    ts.tv_nsec += ms * 1000000L; ts.tv_sec += ts.tv_nsec / 1000000000L; ts.tv_nsec %= 1000000000L;
    return pthread_mutex_timedlock(m, &ts);
}

static void *t1(void *x)
{
    (void)x;
    pthread_mutex_lock(&a); usleep(50000);                 /* держим a, тянемся к b */
    pthread_barrier_wait(&bar);                            /* без барьера под нагрузкой поток 1 успевал взять оба */
    int r = lock_with_timeout(&b, 500);
    printf("поток 1: a взят, b: %s\n", r ? strerror(r) : "взят");
    pthread_barrier_wait(&done);                           /* иначе a освобождался раньше срока потока 2 */
    if (!r) pthread_mutex_unlock(&b);
    pthread_mutex_unlock(&a);
    return NULL;
}

static void *t2(void *x)
{
    (void)x;
    pthread_mutex_lock(&b); usleep(50000);                 /* держим b, тянемся к a — круг замкнулся */
    pthread_barrier_wait(&bar);
    int r = lock_with_timeout(&a, 500);
    printf("поток 2: b взят, a: %s\n", r ? strerror(r) : "взят");
    pthread_barrier_wait(&done);
    if (!r) pthread_mutex_unlock(&a);
    pthread_mutex_unlock(&b);
    return NULL;
}

int main(void)
{
    pthread_t x, y;
    pthread_barrier_init(&bar, NULL, 2);
    pthread_barrier_init(&done, NULL, 2);
    pthread_create(&x, NULL, t1, NULL); pthread_create(&y, NULL, t2, NULL);
    pthread_join(x, NULL); pthread_join(y, NULL);

    pthread_mutexattr_t at; pthread_mutexattr_init(&at);
    pthread_mutexattr_settype(&at, PTHREAD_MUTEX_ERRORCHECK);
    pthread_mutex_t e; pthread_mutex_init(&e, &at);
    pthread_mutex_lock(&e);
    int r = pthread_mutex_lock(&e);                  /* повторный захват тем же потоком */
    printf("ERRORCHECK, второй lock из того же потока: %s\n", strerror(r));
    pthread_mutex_unlock(&e); r = pthread_mutex_unlock(&e);
    printf("ERRORCHECK, лишний unlock: %s\n", strerror(r));
    return 0;
}

cond_queue.c — a job queue on a mutex and a condition variable.

/* Раздел 139: очередь заданий — производитель и три потребителя на мьютексе и условной
   переменной; ожидание в цикле while, ложные пробуждения безопасны. */
#include <stdio.h>
#include <stdlib.h>
#include <pthread.h>
#include <unistd.h>

struct job { struct job *next; int id; };
static struct job *head;
static int done, processed[3];
static pthread_mutex_t qlock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t qready = PTHREAD_COND_INITIALIZER;

static void *consumer(void *arg)
{
    int me = (int)(long)arg;
    for (;;) {
        pthread_mutex_lock(&qlock);
        while (!head && !done)                       /* while, не if: пробуждение не гарантирует данных */
            pthread_cond_wait(&qready, &qlock);      /* отпускает мьютекс на время ожидания */
        if (!head && done) { pthread_mutex_unlock(&qlock); return NULL; }
        struct job *j = head; head = j->next;
        pthread_mutex_unlock(&qlock);
        processed[me]++;                             /* работа — вне мьютекса */
        free(j);
    }
}

int main(void)
{
    pthread_t th[3];
    for (long k = 0; k < 3; k++) pthread_create(&th[k], NULL, consumer, (void *)k);
    for (int i = 0; i < 3000; i++) {
        struct job *j = malloc(sizeof *j); j->id = i;
        pthread_mutex_lock(&qlock);
        j->next = head; head = j;
        pthread_mutex_unlock(&qlock);
        pthread_cond_signal(&qready);                /* разбудить одного */
    }
    pthread_mutex_lock(&qlock); done = 1; pthread_mutex_unlock(&qlock);
    pthread_cond_broadcast(&qready);                 /* разбудить всех: очередь закрыта */
    for (int k = 0; k < 3; k++) pthread_join(th[k], NULL);
    printf("обработано: %d + %d + %d = %d\n", processed[0], processed[1], processed[2],
           processed[0] + processed[1] + processed[2]);
    return 0;
}

Next: section 140 — C11 threads and atomic operations, OpenMP.