src/n_thread_pool.c

Thread pool functions. More...

Structures

	Name
struct	N_PARALLEL_WORKER One worker of a N_PARALLEL_TEAM, as handed to its thread.
struct	N_PARALLEL_TEAM A team of threads kept alive between parallel loops.
struct	N_PARALLEL_SLICE One slice of a parallel loop, as handed to its thread.

Functions

	Name
void *	thread_pool_processing_function(void * param) Internal thread pool processing function.
void *	n_parallel_team_worker(void * arg) Thread entry of a team worker: wait for a loop, run its slice, repeat.
void	n_parallel_team_slice(size_t begin, size_t count, size_t n, size_t k, size_t * out_begin, size_t * out_end) Bounds of slice k of a loop cut in n slices, the same arithmetic n_parallel_for() uses: the first count % n slices get one item more.
void *	n_parallel_slice(void * arg) Thread entry of one slice of a parallel loop.
void	n_parallel_cpu_relax(void ) Hint to the CPU that the caller is spinning.

Attributes

	Name
__thread const struct N_PARALLEL_TEAM *	parallel_team_current The team whose loop the calling thread is running, so a nested loop on the same team runs inline instead of deadlocking on run_lock.
pthread_mutex_t	parallel_lock Guards parallel_budget.
long	parallel_budget get number of core of current system

Defines

	Name
	N_PARALLEL_TEAM_SPIN Polls of the shared state a team thread makes before it sleeps on a condition variable: back-to-back loops (a few microseconds apart) are then picked up without a wake-up through the kernel.

Detailed Description

Thread pool functions.

Author: Castagnier Mickael

Version: 1.0

Date: 30/04/2014

Functions Documentation

function thread_pool_processing_function

void * thread_pool_processing_function(
    void * param
)

Internal thread pool processing function.

Parameters:

• param params of a threaded process

Return: NULL when exiting

function n_parallel_team_worker

static void * n_parallel_team_worker(
    void * arg
)

Thread entry of a team worker: wait for a loop, run its slice, repeat.

Parameters:

• arg the N_PARALLEL_WORKER

Return: NULL

function n_parallel_team_slice

static void n_parallel_team_slice(
    size_t begin,
    size_t count,
    size_t n,
    size_t k,
    size_t * out_begin,
    size_t * out_end
)

Bounds of slice k of a loop cut in n slices, the same arithmetic n_parallel_for() uses: the first count % n slices get one item more.

Parameters:

• begin first index of the loop
• count item count of the loop
• n number of slices
• k slice index
• out_begin receives the first index of the slice
• out_end receives one past its last index

function n_parallel_slice

static void * n_parallel_slice(
    void * arg
)

Thread entry of one slice of a parallel loop.

Parameters:

• arg the N_PARALLEL_SLICE

Return: NULL

function n_parallel_cpu_relax

static inline void n_parallel_cpu_relax(
    void 
)

Hint to the CPU that the caller is spinning.

Attributes Documentation

variable parallel_team_current

static __thread const struct N_PARALLEL_TEAM * parallel_team_current = NULL;

The team whose loop the calling thread is running, so a nested loop on the same team runs inline instead of deadlocking on run_lock.

variable parallel_lock

static pthread_mutex_t parallel_lock = PTHREAD_MUTEX_INITIALIZER;

Guards parallel_budget.

variable parallel_budget

static long parallel_budget = -1;

get number of core of current system

Return: The number of cores or 0 if the system command is not supported

Extra threads n_parallel_for() may have running at once, across every caller; -1 until the first call sizes it from the core count.

Macros Documentation

define N_PARALLEL_TEAM_SPIN

#define N_PARALLEL_TEAM_SPIN 20000

Polls of the shared state a team thread makes before it sleeps on a condition variable: back-to-back loops (a few microseconds apart) are then picked up without a wake-up through the kernel.

Source code

/*
 * Nilorea Library
 * Copyright (C) 2005-2026 Castagnier Mickael
 *
 * Licensed under the Apache License, Version 2.0 (the "License");
 * you may not use this file except in compliance with the License.
 * You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or
 * implied. See the License for the specific language governing
 * permissions and limitations under the License.
 *
 * SPDX-License-Identifier: Apache-2.0
 */


#include <unistd.h>
#include "nilorea/n_common.h"
#include "nilorea/n_log.h"
#include "nilorea/n_thread_pool.h"
#include "nilorea/n_time.h"

#ifdef __linux__
#include <sys/sysinfo.h>
#endif
#include <pthread.h>
#include <string.h>
#include <errno.h>

static long parallel_budget = -1;

static pthread_mutex_t parallel_lock = PTHREAD_MUTEX_INITIALIZER;

typedef struct {
    size_t begin;
    size_t end;
    N_PARALLEL_FN fn;
    void* user;
} N_PARALLEL_SLICE;

static void* n_parallel_slice(void* arg) {
    N_PARALLEL_SLICE* slice = (N_PARALLEL_SLICE*)arg;
    slice->fn(slice->begin, slice->end, slice->user);
    return NULL;
}

int n_parallel_for(size_t begin, size_t end, size_t grain, N_PARALLEL_FN fn, void* user) {
    N_PARALLEL_SLICE slices[64];
    pthread_t threads[64];
    int started[64];
    size_t count = 0, per = 0, at = 0;
    long want = 0, got = 0;
    int n = 0, i = 0;

    if (!fn || end <= begin) return 0;
    if (grain == 0) grain = 1;
    count = end - begin;
    /* how many slices the range is worth, at least a grain each, at most 64 */
    want = (long)(count / grain);
    if (want > 64) want = 64;
    if (want < 2) {
        fn(begin, end, user);
        return 1;
    }
    pthread_mutex_lock(&parallel_lock);
    if (parallel_budget < 0) {
        long cores = get_nb_cpu_cores();
        parallel_budget = cores > 1 ? cores - 1 : 0;
    }
    /* extra threads beyond the caller's own, from what is left of the budget */
    got = want - 1;
    if (got > parallel_budget) got = parallel_budget;
    parallel_budget -= got;
    pthread_mutex_unlock(&parallel_lock);
    if (got <= 0) {
        fn(begin, end, user);
        return 1;
    }
    n = (int)got + 1;
    per = count / (size_t)n;
    /* cleared, so nothing is read before it was written: the joins read only
     * the started handles and n is at least two here, which a static
     * analyser cannot see across the budget's lock */
    memset(threads, 0, sizeof(threads));
    memset(slices, 0, sizeof(slices));
    for (i = 0; i < n; i++) {
        slices[i].begin = begin + at;
        at += per + ((size_t)i < count % (size_t)n ? 1 : 0);
        slices[i].end = begin + at;
        slices[i].fn = fn;
        slices[i].user = user;
        started[i] = 0;
    }
    /* the others on threads, the first on this one; a thread that cannot be
     * started has its slice done here instead */
    for (i = 1; i < n; i++)
        started[i] = pthread_create(&threads[i], NULL, n_parallel_slice, &slices[i]) == 0;
    fn(slices[0].begin, slices[0].end, user);
    for (i = 1; i < n; i++) {
        if (started[i])
            pthread_join(threads[i], NULL);
        else
            fn(slices[i].begin, slices[i].end, user);
    }
    pthread_mutex_lock(&parallel_lock);
    parallel_budget += got;
    pthread_mutex_unlock(&parallel_lock);
    return n;
}

#ifndef N_PARALLEL_TEAM_SPIN
#define N_PARALLEL_TEAM_SPIN 20000
#endif

static inline void n_parallel_cpu_relax(void) {
#if defined(__x86_64__) || defined(__i386__)
    __builtin_ia32_pause();
#elif defined(__aarch64__)
    __asm__ __volatile__("yield");
#endif
}

typedef struct {
    struct N_PARALLEL_TEAM* team;
    size_t index;
    pthread_t thr;
    int started;
} N_PARALLEL_WORKER;

struct N_PARALLEL_TEAM {
    size_t size;
    N_PARALLEL_WORKER* workers;
    pthread_mutex_t run_lock;
    pthread_mutex_t lock;
    pthread_cond_t start;
    pthread_cond_t done;
    unsigned long long generation;
    size_t pending;
    int stopping;
    size_t begin;
    size_t count;
    size_t nb_slices;
    N_PARALLEL_FN fn;
    void* user;
};

static __thread const struct N_PARALLEL_TEAM* parallel_team_current = NULL;

static void n_parallel_team_slice(size_t begin, size_t count, size_t n, size_t k, size_t* out_begin, size_t* out_end) {
    size_t per = count / n, extra = count % n;
    size_t start = k * per + (k < extra ? k : extra);
    *out_begin = begin + start;
    *out_end = *out_begin + per + (k < extra ? 1 : 0);
}

static void* n_parallel_team_worker(void* arg) {
    N_PARALLEL_WORKER* worker = (N_PARALLEL_WORKER*)arg;
    struct N_PARALLEL_TEAM* team = worker->team;
    unsigned long long seen = 0;

    parallel_team_current = team;
    for (;;) {
        size_t begin = 0, count = 0, nb_slices = 0;
        N_PARALLEL_FN fn = NULL;
        void* user = NULL;

        /* poll for the next loop for a while, then sleep until it is published */
        for (int spin = 0; spin < N_PARALLEL_TEAM_SPIN; spin++) {
            if (__atomic_load_n(&team->generation, __ATOMIC_ACQUIRE) != seen || __atomic_load_n(&team->stopping, __ATOMIC_ACQUIRE)) break;
            n_parallel_cpu_relax();
        }
        pthread_mutex_lock(&team->lock);
        while (team->generation == seen && !team->stopping)
            pthread_cond_wait(&team->start, &team->lock);
        if (team->stopping) {
            pthread_mutex_unlock(&team->lock);
            break;
        }
        seen = team->generation;
        begin = team->begin;
        count = team->count;
        nb_slices = team->nb_slices;
        fn = team->fn;
        user = team->user;
        pthread_mutex_unlock(&team->lock);

        if (worker->index < nb_slices) {
            size_t b = 0, e = 0;
            n_parallel_team_slice(begin, count, nb_slices, worker->index, &b, &e);
            fn(b, e, user);
        }

        if (__atomic_sub_fetch(&team->pending, 1, __ATOMIC_ACQ_REL) == 0) {
            /* under the lock, so a caller that found pending > 0 and went to sleep is woken */
            pthread_mutex_lock(&team->lock);
            pthread_cond_signal(&team->done);
            pthread_mutex_unlock(&team->lock);
        }
    }
    return NULL;
}

N_PARALLEL_TEAM* n_parallel_team_new(size_t nb_threads) {
    struct N_PARALLEL_TEAM* team = NULL;
    size_t i = 0;

    if (nb_threads == 0) {
        long cores = get_nb_cpu_cores();
        nb_threads = cores > 0 ? (size_t)cores : 1;
    }
    if (nb_threads > N_PARALLEL_TEAM_MAX) nb_threads = N_PARALLEL_TEAM_MAX;

    team = calloc(1, sizeof(*team));
    if (!team) {
        n_log(LOG_ERR, "could not allocate a parallel team");
        return NULL;
    }
    team->size = nb_threads;
    if (nb_threads > 1) {
        team->workers = calloc(nb_threads - 1, sizeof(*team->workers));
        if (!team->workers) {
            n_log(LOG_ERR, "could not allocate %zu parallel team workers", nb_threads - 1);
            free(team);
            return NULL;
        }
    }
    pthread_mutex_init(&team->run_lock, NULL);
    pthread_mutex_init(&team->lock, NULL);
    pthread_cond_init(&team->start, NULL);
    pthread_cond_init(&team->done, NULL);

    for (i = 0; i + 1 < nb_threads; i++) {
        team->workers[i].team = team;
        team->workers[i].index = i + 1;
        if (pthread_create(&team->workers[i].thr, NULL, n_parallel_team_worker, &team->workers[i]) != 0) {
            n_log(LOG_ERR, "could not start parallel team worker %zu of %zu", i + 1, nb_threads - 1);
            n_parallel_team_free(&team);
            return NULL;
        }
        team->workers[i].started = 1;
    }
    return team;
}

size_t n_parallel_team_size(const N_PARALLEL_TEAM* team) {
    return team ? team->size : 0;
}

int n_parallel_team_run(N_PARALLEL_TEAM* team, size_t begin, size_t end, size_t grain, N_PARALLEL_FN fn, void* user) {
    size_t count = 0, nb_slices = 0, b = 0, e = 0;

    if (!fn || end <= begin) return 0;
    if (grain == 0) grain = 1;
    count = end - begin;
    nb_slices = count / grain;
    if (team && nb_slices > team->size) nb_slices = team->size;
    /* alone, nested inside a loop of this team, or not worth waking anyone */
    if (!team || team->size < 2 || nb_slices < 2 || parallel_team_current == team) {
        fn(begin, end, user);
        return 1;
    }

    pthread_mutex_lock(&team->run_lock);
    pthread_mutex_lock(&team->lock);
    team->begin = begin;
    team->count = count;
    team->nb_slices = nb_slices;
    team->fn = fn;
    team->user = user;
    __atomic_store_n(&team->pending, team->size - 1, __ATOMIC_RELAXED);
    /* release: a worker that sees the new generation sees the loop above */
    __atomic_store_n(&team->generation, team->generation + 1, __ATOMIC_RELEASE);
    pthread_cond_broadcast(&team->start);
    pthread_mutex_unlock(&team->lock);

    parallel_team_current = team;
    n_parallel_team_slice(begin, count, nb_slices, 0, &b, &e);
    fn(b, e, user);
    parallel_team_current = NULL;

    for (int spin = 0; spin < N_PARALLEL_TEAM_SPIN && __atomic_load_n(&team->pending, __ATOMIC_ACQUIRE) > 0; spin++) n_parallel_cpu_relax();
    pthread_mutex_lock(&team->lock);
    while (__atomic_load_n(&team->pending, __ATOMIC_ACQUIRE) > 0)
        pthread_cond_wait(&team->done, &team->lock);
    pthread_mutex_unlock(&team->lock);
    pthread_mutex_unlock(&team->run_lock);
    return (int)nb_slices;
}

void n_parallel_team_free(N_PARALLEL_TEAM** team) {
    size_t i = 0;

    if (!team || !*team) return;
    pthread_mutex_lock(&(*team)->lock);
    __atomic_store_n(&(*team)->stopping, 1, __ATOMIC_RELEASE);
    pthread_cond_broadcast(&(*team)->start);
    pthread_mutex_unlock(&(*team)->lock);
    for (i = 0; i + 1 < (*team)->size; i++)
        if ((*team)->workers[i].started)
            pthread_join((*team)->workers[i].thr, NULL);
    pthread_cond_destroy(&(*team)->start);
    pthread_cond_destroy(&(*team)->done);
    pthread_mutex_destroy(&(*team)->lock);
    pthread_mutex_destroy(&(*team)->run_lock);
    free((*team)->workers);
    free(*team);
    *team = NULL;
}

long int get_nb_cpu_cores() {
    long int nb_procs = 0;
#ifdef __windows__
    SYSTEM_INFO sysinfo;
    GetSystemInfo(&sysinfo);
    nb_procs = (long int)sysinfo.dwNumberOfProcessors;
#else
    nb_procs = sysconf(_SC_NPROCESSORS_ONLN);
#endif
    return nb_procs;
}

long int get_nb_cpu_physical_cores(void) {
#if defined(__windows__) || defined(__linux__)
    long int cores = 0;
#ifdef __windows__
    DWORD len = 0;
    if (!GetLogicalProcessorInformation(NULL, &len) && GetLastError() == ERROR_INSUFFICIENT_BUFFER && len > 0) {
        SYSTEM_LOGICAL_PROCESSOR_INFORMATION* info = malloc(len);
        if (info && GetLogicalProcessorInformation(info, &len)) {
            for (DWORD i = 0; i < len / sizeof(*info); i++)
                if (info[i].Relationship == RelationProcessorCore) cores++;
        }
        free(info);
    }
#elif defined(__linux__)
    /* a core is counted at its first hardware thread: the cpu whose number
     * starts its thread_siblings_list */
    long int n = sysconf(_SC_NPROCESSORS_CONF);
    for (long int i = 0; i < n; i++) {
        char path[128];
        long int first = -1;
        FILE* f = NULL;
        snprintf(path, sizeof(path), "/sys/devices/system/cpu/cpu%ld/topology/thread_siblings_list", i);
        f = fopen(path, "r");
        if (!f) continue;
        if (fscanf(f, "%ld", &first) == 1 && first == i) cores++;
        fclose(f);
    }
#endif
    /* only these two platforms can count cores, so the test lives here: on
     * the others cores would always be 0 (cppcheck knownConditionTrueFalse) */
    if (cores > 0) return cores;
#endif
    return get_nb_cpu_cores();
}

void* thread_pool_processing_function(void* param) {
    THREAD_POOL_NODE* node = (THREAD_POOL_NODE*)param;

    if (!node) {
        n_log(LOG_ERR, "Thread fatal error, no valid payload found, exiting thread function !");
        pthread_exit(NULL);
        return NULL;
    }

    n_log(LOG_DEBUG, "Thread %llu started", (unsigned long long)(uintptr_t)node->thr);

    int thread_state = 0;
    do {
        n_log(LOG_DEBUG, "Thread pool processing func waiting");

        // note: direct procs will automatically post th_start
        sem_wait(&node->th_start);

        pthread_mutex_lock(&node->lock);
        thread_state = node->thread_state;
        pthread_mutex_unlock(&node->lock);

        if (thread_state == RUNNING_THREAD) {
            n_log(LOG_DEBUG, "Thread pool running proc %p", node->func);
            void* (*func_to_run)(void* param) = NULL;
            void* param_to_run = NULL;
            pthread_mutex_lock(&node->lock);
            node->state = RUNNING_PROC;
            func_to_run = node->func;
            param_to_run = node->param;
            pthread_mutex_unlock(&node->lock);

            if (func_to_run) {
                func_to_run(param_to_run);
            }
            n_log(LOG_DEBUG, "Thread pool end proc %p", func_to_run);

            pthread_mutex_lock(&node->lock);
            node->func = NULL;
            node->param = NULL;
            node->state = IDLE_PROC;
            int type = node->type;
            node->type = -1;
            // NORMAL_PROC or DIRECT_PROC do not need to post th_end
            if (type & SYNCED_PROC)
                sem_post(&node->th_end);
            pthread_mutex_unlock(&node->lock);

            refresh_thread_pool(node->thread_pool);
        }
    } while (thread_state != EXITING_THREAD);

    n_log(LOG_DEBUG, "Thread %llu exiting...", (unsigned long long)(uintptr_t)node->thr);

    pthread_mutex_lock(&node->lock);
    node->thread_state = EXITED_THREAD;
    pthread_mutex_unlock(&node->lock);

    n_log(LOG_DEBUG, "Thread %llu exited", (unsigned long long)(uintptr_t)node->thr);

    pthread_exit(NULL);

    return NULL;
} /* thread_pool_processing_function */

THREAD_POOL* new_thread_pool(size_t nbmaxthr, size_t nb_max_waiting) {
    THREAD_POOL* thread_pool = NULL;

    Malloc(thread_pool, THREAD_POOL, 1);
    if (!thread_pool)
        return NULL;

    thread_pool->max_threads = nbmaxthr;
    thread_pool->nb_max_waiting = nb_max_waiting;
    thread_pool->nb_actives = 0;
    thread_pool->discard = NULL;

    thread_pool->thread_list = (THREAD_POOL_NODE**)malloc(nbmaxthr * sizeof(THREAD_POOL_NODE*));
    if (!thread_pool->thread_list) {
        Free(thread_pool);
        return NULL;
    }

    thread_pool->waiting_list = new_generic_list(MAX_LIST_ITEMS);
    if (!thread_pool->waiting_list) {
        n_log(LOG_ERR, "Unable to initialize wait list");
        Free(thread_pool->thread_list);
        Free(thread_pool);
        return NULL;
    }

    pthread_mutex_init(&thread_pool->lock, NULL);

    if (sem_init(&thread_pool->nb_tasks, 0, 1) == -1) {
        int error = errno;
        n_log(LOG_ERR, "sem_init failed : %s on &thread_pool -> nb_tasks", strerror(error));
        list_destroy(&thread_pool->waiting_list);
        pthread_mutex_destroy(&thread_pool->lock);
        Free(thread_pool->thread_list);
        Free(thread_pool);
        return NULL;
    }

    size_t it = 0;
    for (it = 0; it < nbmaxthr; it++) {
        Malloc(thread_pool->thread_list[it], THREAD_POOL_NODE, 1);
        thread_pool->thread_list[it]->type = -1;
        thread_pool->thread_list[it]->state = IDLE_PROC;
        thread_pool->thread_list[it]->thread_state = RUNNING_THREAD;
        thread_pool->thread_list[it]->thread_pool = thread_pool;

        if (sem_init(&thread_pool->thread_list[it]->th_start, 0, 0) == -1) {
            int error = errno;
            n_log(LOG_ERR, "sem_init failed : %s on &thread_pool -> thread_list[ %zu ] -> th_start", strerror(error), it);
            Free(thread_pool->thread_list[it]);
            goto cleanup_error;
        }
        if (sem_init(&thread_pool->thread_list[it]->th_end, 0, 0) == -1) {
            int error = errno;
            n_log(LOG_ERR, "sem_init failed : %s on &thread_pool -> thread_list[ %zu] -> th_end", strerror(error), it);
            sem_destroy(&thread_pool->thread_list[it]->th_start);
            Free(thread_pool->thread_list[it]);
            goto cleanup_error;
        }

        thread_pool->thread_list[it]->func = NULL;
        thread_pool->thread_list[it]->param = NULL;

        pthread_mutex_init(&thread_pool->thread_list[it]->lock, NULL);

        if (pthread_create(&thread_pool->thread_list[it]->thr, NULL, thread_pool_processing_function, (void*)thread_pool->thread_list[it]) != 0) {
            n_log(LOG_ERR, "pthread_create failed : %s for it %zu", strerror(errno), it);
            pthread_mutex_destroy(&thread_pool->thread_list[it]->lock);
            sem_destroy(&thread_pool->thread_list[it]->th_start);
            sem_destroy(&thread_pool->thread_list[it]->th_end);
            Free(thread_pool->thread_list[it]);
            goto cleanup_error;
        }
    }
    return thread_pool;

cleanup_error:
    for (size_t j = 0; j < it; j++) {
        pthread_mutex_lock(&thread_pool->thread_list[j]->lock);
        thread_pool->thread_list[j]->thread_state = EXITING_THREAD;
        sem_post(&thread_pool->thread_list[j]->th_start);
        pthread_mutex_unlock(&thread_pool->thread_list[j]->lock);
        pthread_join(thread_pool->thread_list[j]->thr, NULL);
        pthread_mutex_destroy(&thread_pool->thread_list[j]->lock);
        sem_destroy(&thread_pool->thread_list[j]->th_start);
        sem_destroy(&thread_pool->thread_list[j]->th_end);
        Free(thread_pool->thread_list[j]);
    }
    list_destroy(&thread_pool->waiting_list);
    sem_destroy(&thread_pool->nb_tasks);
    pthread_mutex_destroy(&thread_pool->lock);
    Free(thread_pool->thread_list);
    Free(thread_pool);
    return NULL;
} /* new_thread_pool */

int add_threaded_process(THREAD_POOL* thread_pool, void* (*func_ptr)(void* param), void* param, int mode) {
    if (!thread_pool) {
        n_log(LOG_ERR, "thread_pool is not allocated, can't add processes to it !");
        return FALSE;
    }

    if (!thread_pool->thread_list) {
        n_log(LOG_ERR, "thread_pool thread_list is not allocated, can't add processes to it !");
        return FALSE;
    }

    /* validate that exactly one of NORMAL_PROC, SYNCED_PROC, DIRECT_PROC is set */
    int proc_mode = mode & (NORMAL_PROC | SYNCED_PROC | DIRECT_PROC);
    if (proc_mode != NORMAL_PROC && proc_mode != SYNCED_PROC && proc_mode != DIRECT_PROC) {
        n_log(LOG_ERR, "invalid mode %d: exactly one of NORMAL_PROC, SYNCED_PROC, DIRECT_PROC must be set", mode);
        return FALSE;
    }

    if (!(mode & NO_LOCK)) pthread_mutex_lock(&thread_pool->lock);

    size_t it = 0;
    while (it < thread_pool->max_threads) {
        pthread_mutex_lock(&thread_pool->thread_list[it]->lock);
        if (thread_pool->thread_list[it]->thread_state == RUNNING_THREAD && thread_pool->thread_list[it]->state == IDLE_PROC) {
            break;
        }
        pthread_mutex_unlock(&thread_pool->thread_list[it]->lock);
        it++;
    }
    // we have a free thread slot, and the lock on it
    if (it < thread_pool->max_threads) {
        if (mode & NORMAL_PROC || mode & DIRECT_PROC || mode & SYNCED_PROC) {
            thread_pool->thread_list[it]->func = func_ptr;
            thread_pool->thread_list[it]->param = param;
            thread_pool->thread_list[it]->state = WAITING_PROC;
            thread_pool->thread_list[it]->type = mode;
        } else {
            n_log(LOG_ERR, "unknown mode %d for thread %zu", mode, it);
            pthread_mutex_unlock(&thread_pool->thread_list[it]->lock);
            if (!(mode & NO_LOCK))
                pthread_mutex_unlock(&thread_pool->lock);
            return FALSE;
        }
        if (mode & NORMAL_PROC || mode & DIRECT_PROC)
            sem_post(&thread_pool->thread_list[it]->th_start);
        pthread_mutex_unlock(&thread_pool->thread_list[it]->lock);
        n_log(LOG_DEBUG, "proc %p(%p) added on thread %zu", func_ptr, param, it);
    } else {
        // all thread are occupied -> test waiting lists. not holding thread_list[ it ] lock because it was obligatory unlocked before

        // if already coming from queue, or if it should be part of a synced start, do not re-add && return FALSE
        // it's only an error if SYNCED_PROC mode
        int cancel_and_return = FALSE;
        if (mode & NO_QUEUE) {
            n_log(LOG_DEBUG, "Thread pool active threads are all busy and mode is NO_QUEUE, cannot add %p(%p) to pool %p", func_ptr, param, thread_pool);
            cancel_and_return = TRUE;
        } else if (mode & SYNCED_PROC) {
            n_log(LOG_ERR, "Thread pool active threads are all busy, cannot add SYNCED_PROC %p(%p) to pool %p", func_ptr, param, thread_pool);
            cancel_and_return = TRUE;
        } else if (mode & DIRECT_PROC) {
            n_log(LOG_ERR, "Thread pool active threads are all busy, cannot add DIRECT_PROC %p(%p) to pool %p", func_ptr, param, thread_pool);
            cancel_and_return = TRUE;
        }

        if (cancel_and_return) {
            if (!(mode & NO_LOCK))
                pthread_mutex_unlock(&thread_pool->lock);
            return FALSE;
        }

        // try adding to wait list
        if (thread_pool->nb_max_waiting == 0 || (thread_pool->waiting_list->nb_items < thread_pool->nb_max_waiting)) {
            THREAD_WAITING_PROC* proc = NULL;
            Malloc(proc, THREAD_WAITING_PROC, 1);
            if (!proc) {
                n_log(LOG_ERR, "Failed to allocate THREAD_WAITING_PROC");
                if (!(mode & NO_LOCK)) pthread_mutex_unlock(&thread_pool->lock);
                return FALSE;
            }
            proc->func = func_ptr;
            proc->param = param;
            list_push(thread_pool->waiting_list, proc, free);
            n_log(LOG_DEBUG, "Adding %p %p to waitlist", proc->func, proc->param);
        } else {
            n_log(LOG_ERR, "proc %p(%p) was dropped from waitlist because waitlist of thread pool %p is full", func_ptr, param, thread_pool);
            if (!(mode & NO_LOCK)) pthread_mutex_unlock(&thread_pool->lock);
            return FALSE;
        }
    }

    /* consume idle signal: pool now has work in progress */
    if (sem_trywait(&thread_pool->nb_tasks) != 0 && errno != EAGAIN) {
        int error = errno;
        n_log(LOG_ERR, "sem_trywait nb_tasks error in thread_pool %p: %s", thread_pool, strerror(error));
    }

    if (!(mode & NO_LOCK)) pthread_mutex_unlock(&thread_pool->lock);

    return TRUE;
} /* add_threaded_process */

int start_threaded_pool(THREAD_POOL* thread_pool) {
    if (!thread_pool)
        return FALSE;

    if (!thread_pool->thread_list)
        return FALSE;

    int retval = TRUE;

    pthread_mutex_lock(&thread_pool->lock);
    for (size_t it = 0; it < thread_pool->max_threads; it++) {
        int to_run = 0;
        pthread_mutex_lock(&thread_pool->thread_list[it]->lock);
        if ((thread_pool->thread_list[it]->type & SYNCED_PROC) && thread_pool->thread_list[it]->state == WAITING_PROC) {
            to_run = 1;
        }
        pthread_mutex_unlock(&thread_pool->thread_list[it]->lock);
        if (to_run == 1) {
            if (sem_post(&thread_pool->thread_list[it]->th_start) != 0) {
                int error = errno;
                n_log(LOG_ERR, "sem_post th_start error in thread_pool %p , thread_list[ %zu ] : %s", thread_pool, it, strerror(error));
                retval = FALSE;
            }
        }
    }
    pthread_mutex_unlock(&thread_pool->lock);

    return retval;
} /* start_threaded_pool */

int wait_for_synced_threaded_pool(THREAD_POOL* thread_pool) {
    __n_assert(thread_pool, return FALSE);
    __n_assert(thread_pool->thread_list, return FALSE);

    int retval = TRUE;
    for (size_t it = 0; it < thread_pool->max_threads; it++) {
        int is_synced = 0;
        pthread_mutex_lock(&thread_pool->thread_list[it]->lock);
        if (thread_pool->thread_list[it]->type & SYNCED_PROC) {
            is_synced = 1;
        }
        pthread_mutex_unlock(&thread_pool->thread_list[it]->lock);

        if (is_synced) {
            if (sem_wait(&thread_pool->thread_list[it]->th_end) == -1) {
                int error = errno;
                n_log(LOG_ERR, "sem_wait th_end error in thread_pool %p , thread_list[ %zu ] : %s", thread_pool, it, strerror(error));
                retval = FALSE;
            }
        }
    }
    return retval;
} /* wait_for_synced_threaded_pool */

int wait_for_threaded_pool(THREAD_POOL* thread_pool) {
    __n_assert(thread_pool, return FALSE);
    __n_assert(thread_pool->thread_list, return FALSE);

    /* Poll refresh_thread_pool instead of blocking on nb_tasks.
     *
     * The idle token is consumed by add_threaded_process and posted back ONLY by
     * refresh_thread_pool; a worker finishing its proc does not post it. Refreshing
     * once and then blocking therefore deadlocks whenever the pool is still busy at
     * that moment: refresh sees a RUNNING_PROC thread and posts nothing, the wait
     * blocks, the worker goes idle, and nobody ever signals again. It only appeared
     * to work when the proc happened to finish before that single refresh, so the
     * hang looked timing-dependent and grew more likely the longer the work took.
     *
     * Polling keeps every idleness decision inside refresh_thread_pool (which
     * evaluates it under the pool lock) and cannot miss a wakeup. */
    for (;;) {
        refresh_thread_pool(thread_pool);
        if (sem_trywait(&thread_pool->nb_tasks) == 0) {
            /* restore the idle signal so it can be waited on again or checked */
            sem_post(&thread_pool->nb_tasks);
            return TRUE;
        }
        if (errno != EAGAIN && errno != EINTR) {
            int error = errno;
            n_log(LOG_ERR, "sem_trywait nb_tasks error in thread_pool %p: %s", thread_pool, strerror(error));
            return FALSE;
        }
        u_sleep(N_THREAD_POOL_WAIT_POLL_US);
    }
} /* wait_for_threaded_pool */

int set_thread_pool_discard(THREAD_POOL* thread_pool, void (*discard)(void* param)) {
    __n_assert(thread_pool, return FALSE);
    pthread_mutex_lock(&thread_pool->lock);
    thread_pool->discard = discard;
    pthread_mutex_unlock(&thread_pool->lock);
    return TRUE;
}

int destroy_threaded_pool(THREAD_POOL** pool, unsigned int delay) {
    __n_assert(pool && (*pool), return FALSE);
    __n_assert((*pool)->thread_list, return FALSE);

    int DONE = 0;
    int max_retries = 1000;

    while (!DONE) {
        DONE = 1;
        pthread_mutex_lock(&(*pool)->lock);
        for (size_t it = 0; it < (*pool)->max_threads; it++) {
            pthread_mutex_lock(&(*pool)->thread_list[it]->lock);
            int state = (*pool)->thread_list[it]->state;
            int thread_state = (*pool)->thread_list[it]->thread_state;
            pthread_mutex_unlock(&(*pool)->thread_list[it]->lock);

            if (thread_state == EXITING_THREAD || thread_state == EXITED_THREAD) {
                continue;
            }

            if (state == IDLE_PROC) {
                pthread_mutex_lock(&(*pool)->thread_list[it]->lock);
                (*pool)->thread_list[it]->thread_state = EXITING_THREAD;
                sem_post(&(*pool)->thread_list[it]->th_start);
                pthread_mutex_unlock(&(*pool)->thread_list[it]->lock);
            } else {
                DONE = 0;
            }
        }
        pthread_mutex_unlock(&(*pool)->lock);

        if (!DONE) {
            max_retries--;
            if (max_retries <= 0) {
                pthread_mutex_lock(&(*pool)->lock);
                for (size_t it = 0; it < (*pool)->max_threads; it++) {
                    pthread_mutex_lock(&(*pool)->thread_list[it]->lock);
                    if ((*pool)->thread_list[it]->thread_state != EXITING_THREAD && (*pool)->thread_list[it]->thread_state != EXITED_THREAD) {
                        (*pool)->thread_list[it]->thread_state = EXITING_THREAD;
                        sem_post(&(*pool)->thread_list[it]->th_start);
                    }
                    pthread_mutex_unlock(&(*pool)->thread_list[it]->lock);
                }
                pthread_mutex_unlock(&(*pool)->lock);
                break;
            }
        }

        u_sleep(delay);
    }

    /* join threads without holding pool lock to avoid deadlock:
     * threads may call refresh_thread_pool() which needs pool->lock */
    for (size_t it = 0; it < (*pool)->max_threads; it++) {
        pthread_join((*pool)->thread_list[it]->thr, NULL);
    }

    /* all threads have exited, safe to clean up without locking */
    for (size_t it = 0; it < (*pool)->max_threads; it++) {
        pthread_mutex_destroy(&(*pool)->thread_list[it]->lock);
        sem_destroy(&(*pool)->thread_list[it]->th_start);
        sem_destroy(&(*pool)->thread_list[it]->th_end);
        Free((*pool)->thread_list[it]);
    }
    Free((*pool)->thread_list);

    /* Whatever is still queued was never run, because no thread is left to
     * take it. Said out loud rather than dropped with the list, and handed to
     * the caller's discard function when there is one: the list's own
     * destructor frees the wrapper and knows nothing about the param. */
    if ((*pool)->waiting_list->nb_items > 0) {
        n_log(LOG_ERR, "thread pool %p destroyed with %zu queued job(s) never run%s", (void*)(*pool),
              (*pool)->waiting_list->nb_items,
              (*pool)->discard ? ", handed to its discard function" : ", their params not released");
        if ((*pool)->discard) {
            list_foreach(node, (*pool)->waiting_list) {
                THREAD_WAITING_PROC* proc = (THREAD_WAITING_PROC*)node->ptr;
                if (proc) (*pool)->discard(proc->param);
            }
        }
    }
    list_destroy(&(*pool)->waiting_list);

    sem_destroy(&(*pool)->nb_tasks);

    pthread_mutex_destroy(&(*pool)->lock);

    Free((*pool));

    return TRUE;
} /* destroy_threaded_pool */

int refresh_thread_pool(THREAD_POOL* thread_pool) {
    __n_assert(thread_pool, return FALSE);
    __n_assert(thread_pool->waiting_list, return FALSE);

    /* Trying to empty the wait list */
    int push_status = 0;
    pthread_mutex_lock(&thread_pool->lock);
    if (thread_pool->waiting_list && thread_pool->waiting_list->start)
        push_status = 1;
    while (push_status == 1) {
        LIST_NODE* node = thread_pool->waiting_list->start;
        if (node && node->ptr) {
            THREAD_WAITING_PROC* proc = (THREAD_WAITING_PROC*)node->ptr;
            if (proc) {  // cppcheck-suppress knownConditionTrueFalse ; defensive check after cast
                if (add_threaded_process(thread_pool, proc->func, proc->param, NORMAL_PROC | NO_QUEUE | NO_LOCK) == TRUE) {
                    THREAD_WAITING_PROC* procptr = NULL;
                    LIST_NODE* next_node = node->next;
                    procptr = remove_list_node(thread_pool->waiting_list, node, THREAD_WAITING_PROC);
                    n_log(LOG_DEBUG, "waitlist: adding %p,%p to %p", procptr->func, procptr->param, thread_pool);
                    Free(procptr);
                    (void)next_node; /* advance past removed node (consumed by next iteration) */
                } else {
                    n_log(LOG_DEBUG, "waitlist: cannot add proc %p from waiting list of %p, all active threads are busy !", proc, thread_pool);
                    push_status = 0;
                }
            } else {
                n_log(LOG_ERR, "waitlist: trying to add invalid NULL proc on thread pool %p !", thread_pool);
                push_status = 0;
            }
        } else {
            push_status = 0;
        }
    }  // while( push_status == 1 )

    // update statistics: count threads with assigned work (waiting to run or running)
    thread_pool->nb_actives = 0;
    for (size_t it = 0; it < thread_pool->max_threads; it++) {
        pthread_mutex_lock(&thread_pool->thread_list[it]->lock);
        if (thread_pool->thread_list[it]->state == RUNNING_PROC || thread_pool->thread_list[it]->state == WAITING_PROC)
            thread_pool->nb_actives++;
        pthread_mutex_unlock(&thread_pool->thread_list[it]->lock);
    }

    /* signal idle: no active/waiting threads and waiting list is empty */
    if (thread_pool->nb_actives == 0 && thread_pool->waiting_list && thread_pool->waiting_list->nb_items == 0) {
        /* Clamp the idle token to at most one WITHOUT consulting sem_getvalue.
         * POSIX leaves sem_getvalue's result unspecified while a thread is blocked
         * in sem_wait: an implementation may report zero (glibc) or a negative
         * waiter count (winpthreads does). The old guard posted only when it read
         * exactly 0, so on Windows a waiter blocked in wait_for_threaded_pool made
         * the read non-zero, the post was skipped, and that waiter was stranded for
         * good. Consuming any token already present and then posting exactly one is
         * idempotent and does not depend on that value: with a waiter present the
         * trywait simply fails and the post releases it. */
        if (sem_trywait(&thread_pool->nb_tasks) != 0 && errno != EAGAIN) {
            int error = errno;
            n_log(LOG_ERR, "sem_trywait nb_tasks error in thread_pool %p: %s", thread_pool, strerror(error));
        }
        sem_post(&thread_pool->nb_tasks);
    }

    pthread_mutex_unlock(&thread_pool->lock);

    return TRUE;
}  // refresh_thread_pool()