src/pthreadpool.c

/* Standard C headers */
#include <stdint.h>
#include <stdbool.h>
#include <stdlib.h>
#include <string.h>
#include <assert.h>

/* POSIX headers */
#include <pthread.h>
#include <unistd.h>

/* Library header */
#include <pthreadpool.h>

#define PTHREADPOOL_CACHELINE_SIZE 64
#define PTHREADPOOL_CACHELINE_ALIGNED __attribute__((__aligned__(PTHREADPOOL_CACHELINE_SIZE)))

#if defined(__clang__)
	#if __has_extension(c_static_assert) || __has_feature(c_static_assert)
		#define PTHREADPOOL_STATIC_ASSERT(predicate, message) _Static_assert((predicate), message)
	#else
		#define PTHREADPOOL_STATIC_ASSERT(predicate, message)
	#endif
#elif defined(__GNUC__) && ((__GNUC__ > 4) || (__GNUC__ == 4) && (__GNUC_MINOR__ >= 6))
	/* Static assert is supported by gcc >= 4.6 */
	#define PTHREADPOOL_STATIC_ASSERT(predicate, message) _Static_assert((predicate), message)
#else
	#define PTHREADPOOL_STATIC_ASSERT(predicate, message)
#endif

enum thread_state {
	thread_state_idle,
	thread_state_compute_1d,
	thread_state_shutdown,
};

struct PTHREADPOOL_CACHELINE_ALIGNED thread_info {
	/**
	 * Index of the first element in the work range.
	 * Before processing a new element the owning worker thread increments this value.
	 */
	volatile size_t range_start;
	/**
	 * Index of the element after the last element of the work range.
	 * Before processing a new element the stealing worker thread decrements this value.
	 */
	volatile size_t range_end;
	/**
	 * The number of elements in the work range.
	 * Due to race conditions range_length <= range_end - range_start.
	 * The owning worker thread must decrement this value before incrementing @a range_start.
	 * The stealing worker thread must decrement this value before decrementing @a range_end.
	 */
	volatile size_t range_length;
	/**
	 * The active state of the thread.
	 */
	volatile enum thread_state state;
	/**
	 * Thread number in the 0..threads_count-1 range.
	 */
	size_t thread_number;
	/**
	 * The pthread object corresponding to the thread.
	 */
	pthread_t thread_object;
	/**
	 * Condition variable used to wake up the thread.
	 * When the thread is idle, it waits on this condition variable.
	 */
	pthread_cond_t wakeup_condvar;
};

PTHREADPOOL_STATIC_ASSERT(sizeof(struct thread_info) % PTHREADPOOL_CACHELINE_SIZE == 0, "thread_info structure must occupy an integer number of cache lines (64 bytes)");

struct PTHREADPOOL_CACHELINE_ALIGNED pthreadpool {
	/**
	 * The number of threads that signalled completion of an operation.
	 */
	volatile size_t checkedin_threads;
	/**
	 * The function to call for each item.
	 */
	volatile pthreadpool_function_1d_t function;
	/**
	 * The first argument to the item processing function.
	 */
	void *volatile argument;
	/**
	 * Serializes concurrent calls to @a pthreadpool_compute_* from different threads.
	 */
	pthread_mutex_t execution_mutex;
	/**
	 * Guards access to the @a checkedin_threads variable.
	 */
	pthread_mutex_t barrier_mutex;
	/**
	 * Condition variable to wait until all threads check in.
	 */
	pthread_cond_t barrier_condvar;
	/**
	 * Guards access to the @a state variables.
	 */
	pthread_mutex_t state_mutex;
	/**
	 * Condition variable to wait for change of @a state variable.
	 */
	pthread_cond_t state_condvar;
	/**
	 * The number of threads in the thread pool. Never changes after initialization.
	 */
	size_t threads_count;
	/**
	 * Thread information structures that immediately follow this structure.
	 */
	struct thread_info threads[];
};

PTHREADPOOL_STATIC_ASSERT(sizeof(struct pthreadpool) % PTHREADPOOL_CACHELINE_SIZE == 0, "pthreadpool structure must occupy an integer number of cache lines (64 bytes)");

static void checkin_worker_thread(struct pthreadpool* threadpool) {
	pthread_mutex_lock(&threadpool->barrier_mutex);
	const size_t checkedin_threads = threadpool->checkedin_threads + 1;
	threadpool->checkedin_threads = checkedin_threads;
	if (checkedin_threads == threadpool->threads_count) {
		pthread_cond_signal(&threadpool->barrier_condvar);
	}
	pthread_mutex_unlock(&threadpool->barrier_mutex);
}

static void wait_worker_threads(struct pthreadpool* threadpool) {
	if (threadpool->checkedin_threads != threadpool->threads_count) {
		pthread_mutex_lock(&threadpool->barrier_mutex);
		while (threadpool->checkedin_threads != threadpool->threads_count) {
			pthread_cond_wait(&threadpool->barrier_condvar, &threadpool->barrier_mutex);
		};
		pthread_mutex_unlock(&threadpool->barrier_mutex);
	}
}

static void wakeup_worker_threads(struct pthreadpool* threadpool) {
	pthread_mutex_lock(&threadpool->state_mutex);
	threadpool->checkedin_threads = 0; /* Locking of barrier_mutex not needed: readers are sleeping */
	pthread_cond_broadcast(&threadpool->state_condvar);
	pthread_mutex_unlock(&threadpool->state_mutex); /* Do wake up */
}

inline static bool atomic_decrement(volatile size_t* value) {
	size_t actual_value = *value;
	if (actual_value != 0) {
		size_t expected_value;
		do {
			expected_value = actual_value;
			const size_t new_value = actual_value - 1;
			actual_value = __sync_val_compare_and_swap(value, expected_value, new_value);
		} while ((actual_value != expected_value) && (actual_value != 0));
	}
	return actual_value != 0;
}

static void thread_compute_1d(struct pthreadpool* threadpool, struct thread_info* thread) {
	const pthreadpool_function_1d_t function = threadpool->function;
	void *const argument = threadpool->argument;
	/* Process thread's own range of items */
	size_t range_start = thread->range_start;
	while (atomic_decrement(&thread->range_length)) {
		function(argument, range_start++);
	}
	/* Done, now look for other threads' items to steal */
	const size_t thread_number = thread->thread_number;
	const size_t threads_count = threadpool->threads_count;
	for (size_t tid = (thread_number + 1) % threads_count; tid != thread_number; tid = (tid + 1) % threads_count) {
		struct thread_info* other_thread = &threadpool->threads[tid];
		if (other_thread->state != thread_state_idle) {
			while (atomic_decrement(&other_thread->range_length)) {
				const size_t item_id = __sync_sub_and_fetch(&other_thread->range_end, 1);
				function(argument, item_id);
			}
		}
	}
}

static void* thread_main(void* arg) {
	struct thread_info* thread = (struct thread_info*) arg;
	struct pthreadpool* threadpool = ((struct pthreadpool*) (thread - thread->thread_number)) - 1;

	/* Check in */
	checkin_worker_thread(threadpool);

	/* Monitor the state changes and act accordingly */
	for (;;) {
		/* Lock the state mutex */
		pthread_mutex_lock(&threadpool->state_mutex);
		/* Read the state */
		enum thread_state state;
		while ((state = thread->state) == thread_state_idle) {
			/* Wait for state change */
			pthread_cond_wait(&threadpool->state_condvar, &threadpool->state_mutex);
		}
		/* Read non-idle state */
		pthread_mutex_unlock(&threadpool->state_mutex);
		switch (state) {
			case thread_state_compute_1d:
				thread_compute_1d(threadpool, thread);
				break;
			case thread_state_shutdown:
				return NULL;
			case thread_state_idle:
				/* To inhibit compiler warning */
				break;
		}
		/* Notify the master thread that we finished processing */
		thread->state = thread_state_idle;
		checkin_worker_thread(threadpool);
	};
}

struct pthreadpool* pthreadpool_create(size_t threads_count) {
	if (threads_count == 0) {
		threads_count = (size_t) sysconf(_SC_NPROCESSORS_ONLN);
	}
	struct pthreadpool* threadpool = NULL;
	posix_memalign((void**) &threadpool, 64, sizeof(struct pthreadpool) + threads_count * sizeof(struct thread_info));
	memset(threadpool, 0, sizeof(struct pthreadpool) + threads_count * sizeof(struct thread_info));
	threadpool->threads_count = threads_count;
	pthread_mutex_init(&threadpool->execution_mutex, NULL);
	pthread_mutex_init(&threadpool->barrier_mutex, NULL);
	pthread_cond_init(&threadpool->barrier_condvar, NULL);
	pthread_mutex_init(&threadpool->state_mutex, NULL);
	pthread_cond_init(&threadpool->state_condvar, NULL);

	for (size_t tid = 0; tid < threads_count; tid++) {
		threadpool->threads[tid].thread_number = tid;
		pthread_create(&threadpool->threads[tid].thread_object, NULL, &thread_main, &threadpool->threads[tid]);
	}

	/* Wait until all threads initialize */
	wait_worker_threads(threadpool);
	return threadpool;
}

size_t pthreadpool_get_threads_count(struct pthreadpool* threadpool) {
	return threadpool->threads_count;
}

static inline size_t multiply_divide(size_t a, size_t b, size_t d) {
	#if defined(__SIZEOF_SIZE_T__) && (__SIZEOF_SIZE_T__ == 4)
		return (size_t) (((uint64_t) a) * ((uint64_t) b)) / ((uint64_t) d);
	#elif defined(__SIZEOF_SIZE_T__) && (__SIZEOF_SIZE_T__ == 8)
		return (size_t) (((__uint128_t) a) * ((__uint128_t) b)) / ((__uint128_t) d);
	#else
		#error "Unsupported platform"
	#endif
}

void pthreadpool_compute_1d(
	struct pthreadpool* threadpool,
	pthreadpool_function_1d_t function,
	void* argument,
	size_t items)
{
	/* Protect the global threadpool structures */
	pthread_mutex_lock(&threadpool->execution_mutex);

	/* Spread the work between threads */
	for (size_t tid = 0; tid < threadpool->threads_count; tid++) {
		struct thread_info* thread = &threadpool->threads[tid];
		thread->range_start = multiply_divide(items, tid, threadpool->threads_count);
		thread->range_end = multiply_divide(items, tid + 1, threadpool->threads_count);
		thread->range_length = thread->range_end - thread->range_start;
		thread->state = thread_state_compute_1d;
	}

	/* Setup global arguments */
	threadpool->function = function;
	threadpool->argument = argument;

	/* Wake up the threads */
	wakeup_worker_threads(threadpool);

	/* Wait until the threads finish computation */
	wait_worker_threads(threadpool);

	/* Unprotect the global threadpool structures */
	pthread_mutex_unlock(&threadpool->execution_mutex);
}

void pthreadpool_destroy(struct pthreadpool* threadpool) {
	/* Update threads' states */
	for (size_t tid = 0; tid < threadpool->threads_count; tid++) {
		threadpool->threads[tid].state = thread_state_shutdown;
	}

	/* Wake up the threads */
	wakeup_worker_threads(threadpool);

	/* Wait until all threads return */
	for (size_t tid = 0; tid < threadpool->threads_count; tid++) {
		pthread_join(threadpool->threads[tid].thread_object, NULL);
	}

	/* Release resources */
	pthread_mutex_destroy(&threadpool->execution_mutex);
	pthread_mutex_destroy(&threadpool->barrier_mutex);
	pthread_cond_destroy(&threadpool->barrier_condvar);
	pthread_mutex_destroy(&threadpool->state_mutex);
	pthread_cond_destroy(&threadpool->state_condvar);
	free(threadpool);
}