ABA problem

CAS has been introduced as one of the read-modify-write operations. However, the target object not changing does not necessarily mean that no other threads modified it halfway through. If the target object is changed by another thread and then changed back, the result of comparison would still be equal. In this case, the target object has indeed been modified, yet the operation appears unchanged, compromising its atomicity. We call this ABA problem. Consider the following scenario,

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

atomic_int v = 42;
/* handshake flags: they pin the interleaving down instead of leaving it to
 * the scheduler, as sleeping would.
 */
atomic_bool a_has_read = false, b_done = false;

int threadA(void *args)
{
    (void)args;

    int va = atomic_load(&v);
    printf("A: v = %d\n", va);
    atomic_store(&a_has_read, true);
    /* wait for B; a real program would be preempted here instead */
    while (!atomic_load(&b_done))
        thrd_yield();
    /* v was changed and changed back, so the comparison succeeds and A never
     * learns that v changed at all.
     */
    if (!atomic_compare_exchange_strong(&v, &va, va + 10)) {
        printf("A: CAS failed, v = %d\n", atomic_load(&v));
        return 1;
    }
    printf("A: v = %d\n", atomic_load(&v));

    return 0;
}

int threadB(void *args)
{
    (void)args;

    while (!atomic_load(&a_has_read))
        thrd_yield();

    atomic_fetch_add(&v, 5);
    printf("B: v = %d\n", atomic_load(&v));
    atomic_fetch_sub(&v, 5);
    printf("B: v = %d\n", atomic_load(&v));
    atomic_store(&b_done, true);

    return 0;
}

int main(void)
{
    thrd_t A, B;
    /* Each thread spins on a flag only the other one sets, so an unchecked
     * thrd_create failure would leave the survivor spinning forever.
     */
    if (thrd_create(&A, threadA, NULL) != thrd_success) {
        printf("failed to create thread A.\n");
        return EXIT_FAILURE;
    }
    if (thrd_create(&B, threadB, NULL) != thrd_success) {
        printf("failed to create thread B.\n");
        atomic_store(&b_done, true); /* release A, then reap it */
        thrd_join(A, NULL);
        return EXIT_FAILURE;
    }

    int result_a = 0;
    thrd_join(A, &result_a);
    thrd_join(B, NULL);
    return result_a ? EXIT_FAILURE : EXIT_SUCCESS;
}

The execution result would be:

    A: v = 42
    B: v = 47
    B: v = 42
    A: v = 52

In the example provided, the ABA problem causes thread A to be unaware that variable v has been altered. Since the comparison result indicates that v is unchanged, v + 10 is swapped in. The two flags pin the interleaving down rather than leaving it to the scheduler: a_has_read holds thread B back until A has loaded v, since B finishing first would leave nothing for A to be misled about, and b_done holds A back until B has changed v and changed it back. In a real-world scenario, instead of this handshake, thread A could be paused by a context switch to another task, including preemption by a higher-priority task. This example seems harmless, but things can get nasty when atomic RMW operations are used in more complex data structures.

In a broader context, the ABA problem occurs when changes occur between loading and comparing, but the comparing mechanism is unable to identify that the state of the target object is not the latest, yielding a false positive result.

Returning to the thread pool example in Read-modify-write, it contains the ABA problem as well. In the worker function, we have a thread trying to claim the job.

    job_t *job = atomic_load(&thrd_pool->head->prev);
    ...
    while (job != &thrd_pool->head->job &&
           !atomic_compare_exchange_weak(&thrd_pool->head->prev, &job,
                                         job->prev))
        ;

Consider the following scenario:

1. There is only one job left.
2. Thread A loads the pointer to the job by atomic_load().
3. Thread A is preempted.
4. Thread B claims the job and successfully updates thrd_pool->head->prev.
5. Thread B sets the thread pool state to idle.
6. The main thread finishes waiting and adds more jobs.
7. The memory allocator reuses the recently freed memory for new jobs.
8. Fortunately, the first added job has the same address as the one thread A held.
9. Thread A is back in running state. The comparison result is equal so it updates thrd_pool->head->prev with the old job->prev, which is already a dangling pointer.
10. Another thread loads the dangling pointer from thrd_pool->head->prev.

Notice that even though job->prev is not loaded explicitly before the comparison, the compiler could place loading instructions before the comparison. In the end, the dangling pointer could either point to garbage or trigger a segmentation fault. It could be even worse if a nested ABA problem occurs in thread B. Also, using a memory pool could make allocating a job at the same address more likely, creating more opportunities for the ABA problem to occur. In fact, pre-allocated memory should be used to achieve lock-free behavior since malloc could involve a mutex in a multi-threaded environment.

Being unable to determine whether the target object has been changed through comparison could result in a false positive when the return value of CAS is true. Thus, the atomicity provided by CAS is not guaranteed. The general concept of solving this problem involves adding more information to make different states distinguishable, and then deciding whether to act on the old state or retry with the new state. If acting on the old state is chosen, then safe memory reclamation should be considered as memory may have already been freed by other threads. More aggressively, one might consider a programming paradigm where each operation on the target object has no side effects that modify it. In a later section, we will introduce a different way of implementing atomic RMW operations using LL/SC instructions. The exclusive status provided by LL/SC instructions avoids the pitfall introduced by comparison.

To make different states distinguishable, a common solution is to increment a version number each time the target object is changed. Bundling the target object and version into a comparison ensures that each change marks a distinguishable result. Given a sufficiently large version number, there should be no repeated version numbers. There are multiple methods for storing the version number, depending on the evaluation of the duration before a version number wraps around. In the thread pool example, the target object is a pointer. The unused bits in a pointer can be utilized to store the version number. In addition to embedding the version number into a pointer, we could consider utilizing an additional 32-bit or 64-bit value next to the target object for the version number. This requires the compare-and-swap instruction to be capable of comparing a wider size at once. Sometimes, this is referred to as double-width compare-and-swap. On x86-64 processors, atomic instructions that load or store more than one CPU word require additional hardware support. You can use grep cx16 /proc/cpuinfo to check if the processor supports 16-byte compare-and-swap. For hardware that does not support the desired size, software implementations that may involve locks are used instead, as mentioned in Arbitrarily-sized “atomic” types. Returning to the example, the ABA problem in the following code is fixed by using a version number that increments each time a job is added to the empty queue. On x86-64, add the compiler flag -mcx16 to enable 16-byte compare-and-swap in the worker function.¹

#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 the
     * union off a cache line shared with anything else, and it is also what
     * guarantees the 16-byte alignment a double-width CAS requires. It needs
     * an allocation aligned to match: see tpool_init.
     */
    _Alignas(CACHE_LINE_SIZE) union {
        struct {
            _Atomic(job_t *) prev;
            unsigned long long version;
        };
        _Atomic struct versioned_prev {
            job_t *ptr;
            unsigned long long _version;
        } v_prev;
    };
    char padding[CACHE_LINE_SIZE - sizeof(_Atomic(job_t *)) -
                 sizeof(unsigned long long)]; /* 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 */
            struct versioned_prev job = atomic_load(&thrd_pool->head->v_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.ptr != &thrd_pool->head->job) {
                /* compare 16 byte at once */
                struct versioned_prev next = { .ptr = job.ptr->prev,
                                               ._version = job._version };
                if (atomic_compare_exchange_weak(&thrd_pool->head->v_prev, &job,
                                                 next))
                    break;
            }
            /* worker checks if there is only an idle job in the job queue */
            if (job.ptr == &thrd_pool->head->job) {
                /* worker says it is idle */
                atomic_store(&thrd_pool->state, idle);
                thrd_yield();
                continue;
            }

            job.ptr->future->result =
                (void *)job.ptr->func(job.ptr->future->arg);
            atomic_flag_clear(&job.ptr->future->flag);
            free(job.ptr);
        } 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 double-width CAS on v_prev needs the
     * union 16-byte aligned, and the cache line padding 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;
    idle_job->version = 0ULL;
    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.
     */
    struct versioned_prev cur = atomic_load(&thrd_pool->head->v_prev);
    bool was_empty = cur.ptr == &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) {
        /* Publish the pointer and its version in one 16-byte store. Writing
         * the two halves separately would let a worker observe the new job
         * paired with the old version, which is precisely the window the
         * version number exists to close. This store is unsynchronized: it
         * relies on the employer only adding jobs while the pool is idle,
         * which a worker preempted just after its own state check does not
         * honor. That window is the same one the ABA discussion describes.
         */
        struct versioned_prev next = { .ptr = job,
                                       ._version = cur._version + 1 };
        atomic_store(&thrd_pool->head->v_prev, next);
        /* 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;
}

Notice that, in the struct idle_job, a union is used for type punning, which bundles the pointer and version number for compare-and-swap. Directly casting a job pointer to a pointer that points to a 16-byte object is undefined behavior (due to having different alignment); type punning is used instead. By using this technique, struct idle_job can still be accessed normally in other places, minimizing code modification. Compiler optimizations are conservative on type punning, but it is acceptable for atomic operations. See Atomic fusion. Another way to prevent the ABA problem in the example is to use safe memory reclamation mechanisms. Unlike the previously mentioned approach of acting on the old state, the address of a job is not freed until no thread is using it. This prevents the memory allocator or memory pool from reusing the address and causing problems.

1. When using -mcx16, programs relying on 16-byte atomic operations (e.g., using __int128) may require linking with -latomic, as these operations are implemented via function calls to libatomic on some systems. See https://gcc.gnu.org/bugzilla/show_bug.cgi?id=104688. ↩