Add full thread pool example and description

Approaches are explained first, then the code and explanations on
changes.

Author: idoleat

concurrency-primer.tex

@@ -916,7 +916,7 @@ \subsection{ABA problem}
 
 \begin{ccode}
     job_t *job = atomic_load(&thrd_pool->head->prev);
-    while (!atomic_compare_exchange_strong(&thrd_pool->head->prev, &job,
+    while (!atomic_compare_exchange_weak(&thrd_pool->head->prev, &job,
                                            job->prev)) {
     }
 \end{ccode}
@@ -930,23 +930,46 @@ \subsection{ABA problem}
     \item Thread B sets thread pool state to idle.
     \item Main thread fininshes waiting and adds more jobs.
     \item Memory allocator reuses the recently freed memory as new jobs addresses.
-    \item Fortunately, the first added job has the same address as the one Thread A holded.
+    \item Fortunately, the first added job has the same address as the one thread A holded.
     \item Thread A is back in running state. The comparison result is equal so it updates \monobox{thrd\_pool->head->prev} with the old \monobox{job->prev}, which is already a dangling pointer.
     \item Another thread loads the dangling pointer from \monobox{thrd\_pool->head->prev}.
 \end{enumerate}
 
-Notice that even though \monobox{job->prev} is not loaded explicitly before comparison, compiler could place loading instructions before comparison.
+Notice that even though \monobox{job->prev} is not loaded explicitly before comparison, compiler could place loading instructions before comparison. 
 At the end, the dangling pointer could either point to garbage or trigger segmentation fault.
 It could be even worse if nested ABA problem occurs in thread B.
+Also, the possibility to allocate a job with same address could be higher when using memory pool, meaning that more chances to have ABA problem occurred.
+In fact, pre-allocated memory should be used to achive lock-free since \monobox{malloc} could have mutex involved in multithreaded environment.
 
 Failure to recognize changed target object through comparison can result in stale information.
 The general concept of solving this problem involves adding more information to make different state distinguishable, and then making a decision on 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 the programming paradigm where each operation on the target object does not have a side effect on modifying it.
-In the later section, we will introduce a different way of implementing atomic \textsc{RMW} operations by using LL/SC instructions. The exclusiveness provided by LL/SC instructions avoids the pitfall introduced by comparison.
-
-To make different state distinguishable, a common solution is adding a version number to be compared as well.
-
+In the later section, we will introduce a different way of implementing atomic \textsc{RMW} operations by using LL/SC instructions. The exclusive status provided by LL/SC instructions avoids the pitfall introduced by comparison.
+
+To make different state distinguishable, a common solution is incrementing a version number each time target object is changed.
+By bundling the target object and version into a comparison, it ensures that each change marks a distinguishable result.
+Given a sufficient large size for 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.
+It requires the compare-and-swap instruction to be capable of comparing a wider size at once.
+Sometimes, this is referred to as \introduce{double-width compare-and-swap}.
+On x86-64 processors, for atomic instructions that load or store more that a CPU word size, it needs additional hardware support.
+You can use \monobox{grep cx16 /proc/crpuinfo} to check if the processor supports 16-byte compare-and-swap.
+For hardware that does not support the desired size, software implementations which may have locks involve are used instead as mentioned in \secref{arbitrarily-size}.
+Back to the example, the following code is fixed by using an an version number that increments each time a job is added to the empty queue. On x86-64, add a compiler flag \monobox{-mcx64} to enable 16-byte compare-and-swap in \monobox{worker} function.
+
+\inputminted{c}{./examples/rmw_example_aba.c}
+
+Notice that, in the \monobox{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), thus type punnined is used instead.
+By using this techniques, \monobox{struct idle\_job} still can be accessed normally in other places, minimizing code modification.
+Compiler optimizations are conservative on type punning, but it is acceptable for atomic operations.
+See \secref{fusing}.
+Another way to prevent ABA problem in the example is using safe memory reclamation mechanisms.
+Different from previously mentioned acting on the old state, the address of a job is not freed until no one is using it.
+This prevents memory allocator or memory pool reuses the address and causing problem.
 
 \section{Sequential consistency on weakly-ordered hardware}
 

examples/rmw_example_aba.c

@@ -0,0 +1,192 @@
+#include <stdalign.h>
+#include <stdio.h>
+#include <stdatomic.h>
+#include <threads.h>
+#include <stdlib.h>
+#include <stdbool.h>
+#include <assert.h>
+
+#define CACHE_LINE_SIZE 64
+#define N_JOBS 16
+#define N_THREADS 8
+
+typedef struct job {
+    void *args;
+    struct job *next, *prev;
+} job_t;
+
+typedef struct idle_job {
+    union {
+        struct {
+            _Atomic(job_t *) prev;
+            unsigned long long version;
+        };
+        _Atomic struct B16 {
+            job_t *_prev;
+            unsigned long long _version;
+        } DCAS;
+    };
+    char padding[CACHE_LINE_SIZE - sizeof(_Atomic(job_t *)) -
+                 sizeof(unsigned long long)];
+    job_t job;
+} idle_job_t;
+
+enum state { idle, running, cancelled };
+
+typedef struct thread_pool {
+    atomic_flag initialezed;
+    int size;
+    thrd_t *pool;
+    atomic_int state;
+    thrd_start_t func;
+    // job queue is a SPMC ring buffer
+    idle_job_t *head;
+} thread_pool_t;
+
+static int worker(void *args)
+{
+    if (!args)
+        return EXIT_FAILURE;
+    thread_pool_t *thrd_pool = (thread_pool_t *)args;
+
+    while (1) {
+        if (atomic_load(&thrd_pool->state) == cancelled)
+            return EXIT_SUCCESS;
+        if (atomic_load(&thrd_pool->state) == running) {
+            // claim the job
+            struct B16 job = atomic_load(&thrd_pool->head->DCAS);
+            struct B16 next;
+            do {
+                next._prev = job._prev->prev;
+                next._version = job._version;
+            } while (!atomic_compare_exchange_weak(&thrd_pool->head->DCAS, &job,
+                                                   next));
+
+            if (job._prev->args == NULL) {
+                atomic_store(&thrd_pool->state, idle);
+            } else {
+                printf("Hello from job %d\n", *(int *)job._prev->args);
+                free(job._prev->args);
+                free(job._prev); // could cause dangling pointer in other threads
+            }
+        } else {
+            /* To auto run when jobs added, set status to running if job queue is not empty.
+             * As long as the producer is protected */
+            thrd_yield();
+            continue;
+        }
+    };
+}
+
+static bool thread_pool_init(thread_pool_t *thrd_pool, size_t size)
+{
+    if (atomic_flag_test_and_set(&thrd_pool->initialezed)) {
+        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");
+        return false;
+    }
+
+    // May use memory pool for jobs
+    idle_job_t *idle_job = malloc(sizeof(idle_job_t));
+    if (!idle_job) {
+        printf("Failed to allocate idle job.\n");
+        return false;
+    }
+    // idle_job will always be the first job
+    idle_job->job.args = NULL;
+    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;
+
+    for (size_t i = 0; i < size; i++) {
+        thrd_create(thrd_pool->pool + i, worker, thrd_pool);
+        //TODO: error handling
+    }
+
+    return true;
+}
+
+static void thread_pool_destroy(thread_pool_t *thrd_pool)
+{
+    if (atomic_exchange(&thrd_pool->state, cancelled))
+        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);
+    }
+    while (thrd_pool->head->prev != &thrd_pool->head->job) {
+        job_t *job = thrd_pool->head->prev->prev;
+        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->initialezed);
+}
+
+__attribute__((nonnull(2))) static bool add_job(thread_pool_t *thrd_pool,
+                                                void *args)
+{
+    // May use memory pool for jobs
+    job_t *job = malloc(sizeof(job_t));
+    if (!job)
+        return false;
+
+    // unprotected producer
+    job->args = args;
+    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 (thrd_pool->head->prev == &thrd_pool->head->job) {
+        thrd_pool->head->prev = job;
+        thrd_pool->head->version += 1;
+        // trap worker at idle job
+        thrd_pool->head->job.prev = &thrd_pool->head->job;
+    }
+
+    return true;
+}
+
+static inline void wait_until(thread_pool_t *thrd_pool, int state)
+{
+    while (atomic_load(&thrd_pool->state) != state) {
+        thrd_yield();
+    }
+}
+
+int main()
+{
+    thread_pool_t thrd_pool = { .initialezed = ATOMIC_FLAG_INIT };
+    if (!thread_pool_init(&thrd_pool, N_THREADS)) {
+        printf("failed to init.\n");
+        return 0;
+    }
+    for (int i = 0; i < N_JOBS; i++) {
+        int *id = malloc(sizeof(int));
+        *id = i;
+        add_job(&thrd_pool, id);
+    }
+    // Due to simplified job queue (not protecting producer), starting the pool manually
+    atomic_store(&thrd_pool.state, running);
+    wait_until(&thrd_pool, idle);
+    for (int i = 0; i < N_JOBS; i++) {
+        int *id = malloc(sizeof(int));
+        *id = i;
+        add_job(&thrd_pool, id);
+    }
+    atomic_store(&thrd_pool.state, running);
+    thread_pool_destroy(&thrd_pool);
+    return 0;
+}