THREADS: tools to create and manage a thread pool

Structures

	Name
struct	THREAD_WAITING_PROC Structure of a waiting process item.
struct	THREAD_POOL_NODE
struct	THREAD_POOL Structure of a thread pool.

Types

	Name
typedef void(*)(size_t begin, size_t end, void *user)	N_PARALLEL_FN work done by n_parallel_for() on one slice [begin, end) of a range

Functions

	Name
int	wait_for_threaded_pool(THREAD_POOL * thread_pool) wait for the pool to become idle (no active threads, empty waiting list), blocking without polling
int	wait_for_synced_threaded_pool(THREAD_POOL * thread_pool) wait for all the threads in the pool to terminate processing, blocking but light on the CPU as there is no polling
int	start_threaded_pool(THREAD_POOL * thread_pool) tell all the waiting threads to start their associated process
int	set_thread_pool_discard(THREAD_POOL * thread_pool, void(*)(void *param) discard) set the function destroy_threaded_pool() hands the param of each discarded queued job to
int	refresh_thread_pool(THREAD_POOL * thread_pool) try to add some waiting process on some free thread slots, else do nothing
THREAD_POOL *	new_thread_pool(size_t nbmaxthr, size_t nb_max_waiting) allocate a new thread pool
size_t	n_parallel_team_size(const N_PARALLEL_TEAM * team) Number of participants of a team, the calling thread included.
int	n_parallel_team_run(N_PARALLEL_TEAM * team, size_t begin, size_t end, size_t grain, N_PARALLEL_FN fn, void * user) Run a loop over a range on a team, and return when all of it is done.
N_PARALLEL_TEAM *	n_parallel_team_new(size_t nb_threads) Start a team of threads that stays alive to run many parallel loops.
void	n_parallel_team_free(N_PARALLEL_TEAM ** team) Stop the threads of a team, free it and set the pointer to NULL.
int	n_parallel_for(size_t begin, size_t end, size_t grain, N_PARALLEL_FN fn, void * user) Run a loop over a range on several cores, and return when all of it is done.
long int	get_nb_cpu_physical_cores(void ) number of physical cores (hardware threads sharing a core count once), get_nb_cpu_cores() if unknown
long int	get_nb_cpu_cores() get number of core of current system
int	destroy_threaded_pool(THREAD_POOL ** thread_pool, unsigned int delay) stop every thread and free the pool; queued jobs still waiting when the retry budget runs out are discarded, see set_thread_pool_discard()
int	add_threaded_process(THREAD_POOL * thread_pool, void ()(void *param) func_ptr, void * param, int mode) add a function to run in an available thread inside a pool

Defines

	Name
	WAITING_PROC status of a thread who have proc waiting to be processed
	SYNCED_PROC processing mode for added func, synced start, not queued
	RUNNING_THREAD indicate that the pool is running and ready to use
	RUNNING_PROC status of a thread which proc is currently running
	N_THREAD_POOL_WAIT_POLL_US A thread pool node.
	N_PARALLEL_TEAM_MAX largest number of participants of a N_PARALLEL_TEAM
	NO_QUEUE special processing mode for waiting_list: do not add the work in queue since it' coming from the queue
	NO_LOCK if passed to add_threaded_process, skip main table lock in case we are in a func which is already locking it
	NORMAL_PROC processing mode for added func, synced start, can be queued
	IDLE_PROC status of a thread which is waiting for some proc
	EXITING_THREAD indicate that the pool is exiting, unfinished jobs will finish and the pool will exit the threads and enter the EXITED state
	EXITED_THREAD indicate that the pool is off, all jobs have been consumed
	DIRECT_PROC processing mode for added func, direct start, not queued

Types Documentation

typedef N_PARALLEL_FN

typedef void(* N_PARALLEL_FN) (size_t begin, size_t end, void *user);

work done by n_parallel_for() on one slice [begin, end) of a range

Functions Documentation

function wait_for_threaded_pool

int wait_for_threaded_pool(
    THREAD_POOL * thread_pool
)

wait for the pool to become idle (no active threads, empty waiting list), blocking without polling

Parameters:

• thread_pool The thread pool to wait on

Return: TRUE or FALSE

Wait for the thread pool to become idle (no active threads, empty waiting list), blocking without polling.

wait for the pool to become idle (no active threads, empty waiting list), blocking without polling

function wait_for_synced_threaded_pool

int wait_for_synced_threaded_pool(
    THREAD_POOL * thread_pool
)

wait for all the threads in the pool to terminate processing, blocking but light on the CPU as there is no polling

Parameters:

• thread_pool The thread pool to wait

Return: TRUE or FALSE

wait for all the launched process, blocking but light on the CPU as there is no polling

wait for all the threads in the pool to terminate processing, blocking but light on the CPU as there is no polling

function start_threaded_pool

int start_threaded_pool(
    THREAD_POOL * thread_pool
)

tell all the waiting threads to start their associated process

Parameters:

• thread_pool The thread pool to launch

Return: TRUE or FALSE

Launch the process waiting for execution in the thread pool.

tell all the waiting threads to start their associated process

function set_thread_pool_discard

int set_thread_pool_discard(
    THREAD_POOL * thread_pool,
    void(*)(void *param) discard
)

set the function destroy_threaded_pool() hands the param of each discarded queued job to

Parameters:

• thread_pool The pool. Must not be NULL.
• discard Called once per discarded param, from the thread calling destroy_threaded_pool(), after every worker has been joined. NULL to leave discarded params alone, which is the default.

Return: TRUE, or FALSE if thread_pool is NULL

Set what destroy_threaded_pool() does with the param of a queued job it discards.

set the function destroy_threaded_pool() hands the param of each discarded queued job to

The pool never owns a job's param and cannot know how to release one. A job still in the waiting list when destroy_threaded_pool() gives up waiting is not run, so nothing else will ever see its param; this callback is the one place that can release it. Without one those params are left alone, which is a leak whenever the caller allocated them.

function refresh_thread_pool

int refresh_thread_pool(
    THREAD_POOL * thread_pool
)

try to add some waiting process on some free thread slots, else do nothing

Parameters:

• thread_pool The thread pool to refresh

Return: TRUE or FALSE

try to add some waiting DIRECT_PROCs on some free thread slots, else do nothing

try to add some waiting process on some free thread slots, else do nothing

function new_thread_pool

THREAD_POOL * new_thread_pool(
    size_t nbmaxthr,
    size_t nb_max_waiting
)

allocate a new thread pool

Parameters:

• nbmaxthr number of active threads in the pool
• nb_max_waiting max number of waiting procs in the pool. Zero for no limit

Return: NULL or a new thread pool object

Create a new pool of nbmaxthr threads.

allocate a new thread pool

function n_parallel_team_size

size_t n_parallel_team_size(
    const N_PARALLEL_TEAM * team
)

Number of participants of a team, the calling thread included.

Parameters:

• team the team

Return: the participant count, 0 if team is NULL

function n_parallel_team_run

int n_parallel_team_run(
    N_PARALLEL_TEAM * team,
    size_t begin,
    size_t end,
    size_t grain,
    N_PARALLEL_FN fn,
    void * user
)

Run a loop over a range on a team, and return when all of it is done.

Parameters:

• team the team; NULL runs the whole range on the calling thread
• begin first index
• end one past the last index
• grain smallest slice worth a participant of its own; 0 is taken as 1
• fn the work for one slice
• user passed to every call of fn

Return: the number of slices it was run in, at least 1 for a non empty range, 0 otherwise

The range [begin, end) is cut into at most n_parallel_team_size() contiguous slices of at least grain items, and fn is called once per slice, the calling thread doing the first one. The cut depends only on the range, the grain and the team size, never on timing, and slice k is always run by participant k, so a computation that reduces per slice gives the same result on every run.

Several threads may share a team: their loops run one after the other. A loop started from inside fn of a loop of the same team runs on the calling thread alone instead of deadlocking.

function n_parallel_team_new

N_PARALLEL_TEAM * n_parallel_team_new(
    size_t nb_threads
)

Start a team of threads that stays alive to run many parallel loops.

Parameters:

• nb_threads number of participants in a loop, the calling thread included, so nb_threads - 1 threads are started. 0 takes the number of cores. 1 is valid and starts no thread: every loop then runs on the caller. Capped at N_PARALLEL_TEAM_MAX.

Return: the team, or NULL on failure (no thread is left running)

n_parallel_for() starts and joins its threads on every call, which costs tens of microseconds per thread: nothing for a loop that runs for milliseconds, everything for a program that runs thousands of short loops per second. A team pays for its threads once; between loops they poll for the next one for a short while (tens of microseconds to a fraction of a millisecond, so back-to-back loops start without a kernel wake-up), then sleep on a condition variable and use no CPU. The caller waits for the end of a loop the same way.

function n_parallel_team_free

void n_parallel_team_free(
    N_PARALLEL_TEAM ** team
)

Stop the threads of a team, free it and set the pointer to NULL.

Parameters:

• team address of the team; NULL or a NULL team is a no-op

Must not be called while a loop is running on the team.

function n_parallel_for

int n_parallel_for(
    size_t begin,
    size_t end,
    size_t grain,
    N_PARALLEL_FN fn,
    void * user
)

Run a loop over a range on several cores, and return when all of it is done.

Parameters:

• begin first index
• end one past the last index
• grain smallest slice worth a thread of its own; 0 is taken as 1
• fn the work for one slice
• user passed to every call of fn

Return: the number of slices it was run in, at least 1 for a non empty range, 0 otherwise

The range [begin, end) is cut into contiguous slices, at least grain items each, and fn is called once per slice, the calling thread doing one of them. Slices never overlap, so a loop whose iterations write only their own item needs no lock.

The extra threads come from one budget shared by the whole process: the cores less one. A caller that finds it spent - because many workers of a pool are each running a parallel loop - gets fewer threads or none, and runs the rest itself, so nested and concurrent use never oversubscribes the machine. A range smaller than two grains runs on the calling thread alone, with no thread started.

function get_nb_cpu_physical_cores

long int get_nb_cpu_physical_cores(
    void 
)

number of physical cores (hardware threads sharing a core count once), get_nb_cpu_cores() if unknown

Return: the physical core count, or get_nb_cpu_cores() when the topology cannot be read

Number of physical cores of the system: hardware threads that share a core (SMT, hyper-threading) count once.

number of physical cores (hardware threads sharing a core count once), get_nb_cpu_cores() if unknown

Memory-bound loops often run faster on one thread per core than on every hardware thread.

function get_nb_cpu_cores

long int get_nb_cpu_cores()

get number of core of current system

function destroy_threaded_pool

int destroy_threaded_pool(
    THREAD_POOL ** thread_pool,
    unsigned int delay
)

stop every thread and free the pool; queued jobs still waiting when the retry budget runs out are discarded, see set_thread_pool_discard()

Parameters:

• pool The THREAD_POOL *object to kill; set to NULL on return
• delay time in usec between each check

Return: TRUE or FALSE

Stop every thread of a pool and free it.

stop every thread and free the pool; queued jobs still waiting when the retry budget runs out are discarded, see set_thread_pool_discard()

Retries every delay microseconds, up to 1000 times, for each busy thread to go idle. A job finishing hands its thread the next queued job, so while the budget lasts the waiting list drains. When it runs out, every thread is told to exit and destroy joins them, which still waits for the job each is running. What running out of budget changes is only this: the jobs still in the waiting list are not run. They are counted in the log at LOG_ERR and, if set_thread_pool_discard() gave one, each param is handed to the discard function. Call wait_for_threaded_pool() first to have every queued job run.

function add_threaded_process

int add_threaded_process(
    THREAD_POOL * thread_pool,
    void *(*)(void *param) func_ptr,
    void * param,
    int mode
)

add a function to run in an available thread inside a pool

Parameters:

• thread_pool The target thread pool
• func_ptr The function pointer to launch
• param Eventual parameter struct to pass to the function
• mode NORMAL_PROC: add to an available active thread or queue in the waiting list. SYNCED_PROC: add to an available thread and wait for a start call, no waiting list. DIRECT_PROC: add to an available thread, no waiting list.

Return: TRUE or FALSE

add a function and params to a thread pool

add a function to run in an available thread inside a pool

Macros Documentation

define WAITING_PROC

#define WAITING_PROC 32

status of a thread who have proc waiting to be processed

define SYNCED_PROC

#define SYNCED_PROC 2

processing mode for added func, synced start, not queued

define RUNNING_THREAD

#define RUNNING_THREAD 128

indicate that the pool is running and ready to use

define RUNNING_PROC

#define RUNNING_PROC 64

status of a thread which proc is currently running

define N_THREAD_POOL_WAIT_POLL_US

#define N_THREAD_POOL_WAIT_POLL_US 1000

A thread pool node.

microseconds wait_for_threaded_pool() sleeps between pool-idleness checks

define N_PARALLEL_TEAM_MAX

#define N_PARALLEL_TEAM_MAX 256

largest number of participants of a N_PARALLEL_TEAM

define NO_QUEUE

#define NO_QUEUE 8

special processing mode for waiting_list: do not add the work in queue since it' coming from the queue

define NO_LOCK

#define NO_LOCK 1024

if passed to add_threaded_process, skip main table lock in case we are in a func which is already locking it

define NORMAL_PROC

#define NORMAL_PROC 1

processing mode for added func, synced start, can be queued

define IDLE_PROC

#define IDLE_PROC 16

status of a thread which is waiting for some proc

define EXITING_THREAD

#define EXITING_THREAD 256

indicate that the pool is exiting, unfinished jobs will finish and the pool will exit the threads and enter the EXITED state

define EXITED_THREAD

#define EXITED_THREAD 512

indicate that the pool is off, all jobs have been consumed

define DIRECT_PROC

#define DIRECT_PROC 4

processing mode for added func, direct start, not queued