worker.c

#define _POSIX_C_SOURCE 200809L

#include "worker.h"
#include <lektor/common.h>

#include <stdio.h>
#include <pthread.h>
#include <assert.h>
#include <sys/sysinfo.h>
#include <sched.h>
#include <errno.h>
#include <string.h>

static void *
__worker_thread(void *__pool)
{
    struct worker_pool *pool = (struct worker_pool *) __pool;
    volatile void *volatile arg;
    worker_function func;
    for (;;) {
        errno = 0;
        if (pthread_mutex_lock(&pool->lock)) {
            LOG_ERROR("WORKER", "Failed to lock mutex: %s", strerror(errno));
            abort();
        }
        if (pool->len) {
            --(pool->len);
            ++(pool->thread_working);
            func = pool->functions[pool->len];
            arg  = pool->args[pool->len];
        } else if (pool->exit) {
            pthread_mutex_unlock(&pool->lock);
            LOG_INFO("WORKER", "Exiting");
            break;
        } else {
            func = NULL;
            pthread_mutex_unlock(&pool->lock);
            sched_yield();
            continue;
        }

        pthread_mutex_unlock(&pool->lock);
        func((void *) arg);

        while (pthread_mutex_trylock(&pool->lock))
            sched_yield();
        --(pool->thread_working);
        pthread_mutex_unlock(&pool->lock);
    }
    return NULL;
}

int
worker_pool_new(struct worker_pool *ret, size_t init_size, size_t thread_count)
{
    if (!thread_count)
        thread_count = get_nprocs_conf();
    struct worker_pool __ret = {
        .functions   = malloc(sizeof(worker_function) * init_size),
        .args        = malloc(sizeof(void *) * init_size),
        .threads     = malloc(sizeof(pthread_t) * thread_count),
        .size        = init_size,
        .thread_size = thread_count,
        .len         = 0u,
        .exit        = 0,
    };
    if (!__ret.functions || !__ret.args || !__ret.threads) {
        LOG_ERROR("WORKER", "Out of memory");
        __ret.thread_size = 0u;
        worker_pool_free(&__ret);