Keyboard shortcuts

Press ← or → to navigate between chapters

Press S or / to search in the book

Press ? to show this help

Press Esc to hide this help

Example

The following example code is a simplified implementation of a thread pool, which demonstrates the use of the C11 atomic library.

#include <stdio.h>
#include <stdatomic.h>
#include <threads.h>
#include <stdlib.h>
#include <stdbool.h>
#include <assert.h>
#include <math.h>

#define PRECISION 100 /* upper bound in BPP sum */
#define CACHE_LINE_SIZE 64
#define N_THREADS 64

struct tpool_future {
    void *result;
    void *arg;
    atomic_flag flag;
};

typedef struct job {
    void *(*func)(void *);
    struct tpool_future *future;
    struct job *next, *prev;
} job_t;

typedef struct idle_job {
    /* Padding alone only sizes the struct; the alignment is what keeps "prev"
     * off a cache line shared with anything else, and it requires an
     * allocation aligned to match. See tpool_init.
     */
    _Alignas(CACHE_LINE_SIZE) _Atomic(job_t *) prev;
    char padding[CACHE_LINE_SIZE -
                 sizeof(_Atomic(job_t *))]; /* avoid false sharing */
    job_t job;
} idle_job_t;

enum state { idle, running, cancelled };

typedef struct tpool {
    atomic_flag initialized;
    int size;
    thrd_t *pool;
    atomic_int state;
    thrd_start_t func;
    idle_job_t *head; /* job queue is a SPMC ring buffer */
} tpool_t;

static struct tpool_future *tpool_future_create(void *arg)
{
    struct tpool_future *future = malloc(sizeof(struct tpool_future));
    if (future) {
        future->result = NULL;
        future->arg = arg;
        atomic_flag_clear(&future->flag);
        atomic_flag_test_and_set(&future->flag);
    }
    return future;
}

void tpool_future_wait(struct tpool_future *future)
{
    while (atomic_flag_test_and_set(&future->flag))
        ;
}

void tpool_future_destroy(struct tpool_future *future)
{
    free(future->result);
    free(future);
}

static int worker(void *args)
{
    if (!args)
        return EXIT_FAILURE;
    tpool_t *thrd_pool = (tpool_t *)args;

    while (1) {
        /* worker is laid off */
        if (atomic_load(&thrd_pool->state) == cancelled)
            return EXIT_SUCCESS;
        if (atomic_load(&thrd_pool->state) == running) {
            /* worker takes the job */
            job_t *job = atomic_load(&thrd_pool->head->prev);
            /* A failed compare-exchange reloads "job", so the idle job has to
             * be ruled out on every iteration, not just once up front.
             */
            while (job != &thrd_pool->head->job &&
                   !atomic_compare_exchange_weak(&thrd_pool->head->prev, &job,
                                                 job->prev))
                ;
            /* worker checks if there is only an idle job in the job queue */
            if (job == &thrd_pool->head->job) {
                /* worker says it is idle */
                atomic_store(&thrd_pool->state, idle);
                thrd_yield();
                continue;
            }
            job->future->result = (void *)job->func(job->future->arg);
            atomic_flag_clear(&job->future->flag);
            free(job);
        } else {
            /* worker is idle */
            thrd_yield();
        }
    }
    return EXIT_SUCCESS;
}

static bool tpool_init(tpool_t *thrd_pool, size_t size)
{
    if (atomic_flag_test_and_set(&thrd_pool->initialized)) {
        printf("This thread pool has already been initialized.\n");
        return false;
    }

    assert(size > 0);
    thrd_pool->pool = malloc(sizeof(thrd_t) * size);
    if (!thrd_pool->pool) {
        printf("Failed to allocate thread identifiers.\n");
        /* release the claim, otherwise the pool can never be initialized */
        atomic_flag_clear(&thrd_pool->initialized);
        return false;
    }

    /* aligned_alloc, not malloc: the cache line padding in idle_job_t is only
     * worth anything if the allocation starts on a cache line boundary.
     */
    idle_job_t *idle_job =
        aligned_alloc(_Alignof(idle_job_t), sizeof(idle_job_t));
    if (!idle_job) {
        printf("Failed to allocate idle job.\n");
        free(thrd_pool->pool);
        atomic_flag_clear(&thrd_pool->initialized);
        return false;
    }

    /* idle_job will always be the first job */
    idle_job->job.next = &idle_job->job;
    idle_job->job.prev = &idle_job->job;
    idle_job->prev = &idle_job->job;
    thrd_pool->func = worker;
    thrd_pool->head = idle_job;
    thrd_pool->state = idle;
    thrd_pool->size = size;

    /* employer hires many workers */
    for (size_t i = 0; i < size; i++) {
        if (thrd_create(thrd_pool->pool + i, worker, thrd_pool) !=
            thrd_success) {
            printf("Failed to create worker %zu.\n", i);
            /* lay off whoever was already hired before giving up */
            atomic_store(&thrd_pool->state, cancelled);
            while (i--)
                thrd_join(thrd_pool->pool[i], NULL);
            free(idle_job);
            free(thrd_pool->pool);
            /* init undoes itself completely, so there is nothing left for
             * tpool_destroy to reclaim and the caller must not call it.
             * Clear the fields anyway, so a later tpool_init has no stale
             * pointer or count to trip over.
             */
            thrd_pool->pool = NULL;
            thrd_pool->head = NULL;
            thrd_pool->size = 0;
            atomic_flag_clear(&thrd_pool->initialized);
            return false;
        }
    }

    return true;
}

static void tpool_destroy(tpool_t *thrd_pool)
{
    if (atomic_exchange(&thrd_pool->state, cancelled) == running)
        printf("Thread pool cancelled with jobs still running.\n");

    for (int i = 0; i < thrd_pool->size; i++)
        thrd_join(thrd_pool->pool[i], NULL);

    /* Workers are all joined, so the queue is ours alone now. Unclaimed jobs
     * own a future that nobody will ever wait on; free both.
     */
    while (thrd_pool->head->prev != &thrd_pool->head->job) {
        job_t *job = thrd_pool->head->prev->prev;
        tpool_future_destroy(thrd_pool->head->prev->future);
        free(thrd_pool->head->prev);
        thrd_pool->head->prev = job;
    }
    free(thrd_pool->head);
    free(thrd_pool->pool);
    atomic_fetch_and(&thrd_pool->state, 0);
    atomic_flag_clear(&thrd_pool->initialized);
}

/* Use the Bailey–Borwein–Plouffe formula to approximate PI */
static void *bbp(void *arg)
{
    int k = *(int *)arg;
    double sum = (4.0 / (8 * k + 1)) - (2.0 / (8 * k + 4)) -
                 (1.0 / (8 * k + 5)) - (1.0 / (8 * k + 6));
    double *product = malloc(sizeof(double));
    if (!product)
        return NULL;

    *product = 1 / pow(16, k) * sum;
    return (void *)product;
}

struct tpool_future *add_job(tpool_t *thrd_pool, void *(*func)(void *),
                             void *arg)
{
    job_t *job = malloc(sizeof(job_t));
    if (!job)
        return NULL;

    struct tpool_future *future = tpool_future_create(arg);
    if (!future) {
        free(job);
        return NULL;
    }

    /* Workers pop from the back and free as they go, but nothing updates the
     * front link on the way, so once the queue has drained head->job.next
     * still names the last job freed. Drop it before linking, otherwise the
     * writes below land in freed memory.
     */
    bool was_empty = thrd_pool->head->prev == &thrd_pool->head->job;
    if (was_empty)
        thrd_pool->head->job.next = &thrd_pool->head->job;

    job->func = func;
    job->future = future;
    job->next = thrd_pool->head->job.next;
    job->prev = &thrd_pool->head->job;
    thrd_pool->head->job.next->prev = job;
    thrd_pool->head->job.next = job;
    if (was_empty) {
        thrd_pool->head->prev = job;
        /* the previous job of the idle job is itself */
        thrd_pool->head->job.prev = &thrd_pool->head->job;
    }
    return future;
}

static inline void wait_until(tpool_t *thrd_pool, int state)
{
    while (atomic_load(&thrd_pool->state) != state)
        thrd_yield();
}

int main(void)
{
    int bbp_args[PRECISION];
    struct tpool_future *futures[PRECISION];
    double bbp_sum = 0;

    tpool_t thrd_pool = { .initialized = ATOMIC_FLAG_INIT };
    if (!tpool_init(&thrd_pool, N_THREADS)) {
        printf("failed to init.\n");
        return EXIT_FAILURE;
    }
    /* employer asks workers to work */
    atomic_store(&thrd_pool.state, running);

    /* employer waits ... until workers are idle */
    wait_until(&thrd_pool, idle);

    /* employer adds more jobs to the job queue */
    for (int i = 0; i < PRECISION; i++) {
        bbp_args[i] = i;
        futures[i] = add_job(&thrd_pool, bbp, &bbp_args[i]);
        if (!futures[i]) {
            printf("Failed to add job %d.\n", i);
            /* Jobs handed out before this point own a future each, and a
             * worker that claims one frees the job but not the future. Let
             * them drain so those futures can be reclaimed here.
             */
            atomic_store(&thrd_pool.state, running);
            for (int j = 0; j < i; j++) {
                tpool_future_wait(futures[j]);
                tpool_future_destroy(futures[j]);
            }
            /* the pool is drained, so let it say so */
            wait_until(&thrd_pool, idle);
            tpool_destroy(&thrd_pool);
            return EXIT_FAILURE;
        }
    }

    /* employer asks workers to work */
    atomic_store(&thrd_pool.state, running);

    /* employer waits for the result of the job */
    bool complete = true;
    for (int i = 0; i < PRECISION; i++) {
        tpool_future_wait(futures[i]);
        /* bbp returns NULL if it could not allocate its result */
        if (futures[i]->result)
            bbp_sum += *(double *)(futures[i]->result);
        else {
            printf("Job %d produced no result.\n", i);
            complete = false;
        }
        tpool_future_destroy(futures[i]);
    }

    /* employer destroys the job queue and lays workers off. Wait for the
     * workers to park first: tpool_destroy reports on a pool it cancels
     * while running, and a completed future does not by itself mean the
     * last worker has published idle.
     */
    wait_until(&thrd_pool, idle);
    tpool_destroy(&thrd_pool);
    printf("PI calculated with %d terms: %.15f\n", PRECISION, bbp_sum);
    return complete ? EXIT_SUCCESS : EXIT_FAILURE;
}

Stdout of the program is:

PI calculated with 100 terms: 3.141592653589793

Exchange In function tpool_destroy, atomic_exchange(&thrd_pool->state, cancelled) reads the current state and replaces it with “cancelled”. A warning message is printed if the pool is destroyed while workers are still “running”. If the exchange is not performed atomically, we may initially get the state as “running”. Subsequently, a thread could set the state to “cancelled” after finishing the last one, resulting in a false warning.

Test and set In this example, the scenario is as follows: First, the main thread initially acquires a lock future->flag and then sets it true, which is akin to creating a job and then transferring its ownership to the worker. Subsequently, the main thread will be blocked until the worker clears the flag. This indicates that the main thread will wait until the worker completes the job and returns ownership to the main thread, allowing the two threads to cooperate correctly.

Fetch and… In the function tpool_destroy, atomic_fetch_and is utilized as a means to set the state to “idle”. Yet, in this case, it is not necessary, as the pool needs to be reinitialized for further use regardless. Its return value could be further utilized, for instance, to report the previous state and perform additional actions.

Compare and swap Once threads are created in the thread pool as workers, they will continuously search for jobs to do. Jobs are taken from the tail of the job queue. To take a job without being taken by another worker halfway through, we need to atomically change the pointer to the last job. Otherwise, the last job is under race. The while loop in the function worker,

while (job != &thrd_pool->head->job &&
       !atomic_compare_exchange_weak(&thrd_pool->head->prev, &job,
                                     job->prev)) {
}

, keeps trying to claim the job atomically until it succeeds. A failed compare-exchange writes the current value back into job, so the guard is re-evaluated on every iteration: another worker may have taken the last job in the meantime, leaving the idle job, which must never be claimed.

Built-in post increment and decrement operators and compound assignment on atomic objects, such as ++ and +=, are read-modify-write atomic operations with total sequentially consistent ordering as well. They behave equivalently to a do while loop. See C11 standard 6.5.2.4 and 6.5.16.2 for more details.

What if claiming a job, which updates thrd_pool->head->prev, is not done atomically? Two or more threads could have races updating thrd_pool->head->prev and working on the same job. Data races are undefined behavior in C11 and C++11. Working on the same job can lead to duplication of the calculation of job->future->result, use after free and double free on the job.

But even when jobs are claimed atomically, a thread can still end up holding a job that has been freed. This is a defect of the example code. Jobs in the example are dynamically allocated. They are freed after a worker finishes each job. However, this situation may lead to dangling pointers for workers that are still holding and attempting to claim the job. If jobs are intended to be dynamically allocated, then safe memory reclamation should be implemented for such shared objects. RCU, hazard pointers, and reference counting are major ways of solving this problem.