threadpool.c

#include "threadpool.h"
#include <stdlib.h>

static void* worker(void *p) {
    threadpool_t *tp = (threadpool_t*)p;
    for (;;) {
        pthread_mutex_lock(&tp->mtx);
        while (!tp->stop && tp->size == 0)
            pthread_cond_wait(&tp->cv_nonempty, &tp->mtx);
        if (tp->stop && tp->size == 0) {
            pthread_mutex_unlock(&tp->mtx);
            break;
        }
        tp_task *t = tp->head;
        tp->head = t->next;
        if (!tp->head) tp->tail = NULL;
        tp->size--;
        pthread_cond_signal(&tp->cv_nonfull);
        pthread_mutex_unlock(&tp->mtx);
        t->fn(t->arg);
        free(t);
    }
    return NULL;
}

int tp_init(threadpool_t *tp, int nthreads, int cap) {
    *tp = (threadpool_t){0};
    pthread_mutex_init(&tp->mtx, NULL);
    pthread_cond_init(&tp->cv_nonempty, NULL);
    pthread_cond_init(&tp->cv_nonfull, NULL);
    tp->cap = cap;
    tp->nthreads = nthreads;
    tp->threads = (pthread_t*)calloc(nthreads, sizeof(pthread_t));
    if (!tp->threads) return -1;
    for (int i = 0; i < nthreads; ++i)
        pthread_create(&tp->threads[i], NULL, worker, tp);
    return 0;
}

void tp_submit(threadpool_t *tp, tp_task_fn fn, void *arg) {
    tp_task *t = (tp_task*)malloc(sizeof(tp_task));
    t->fn = fn; t->arg = arg; t->next = NULL;
    pthread_mutex_lock(&tp->mtx);
    while (!tp->stop && tp->size >= tp->cap)
        pthread_cond_wait(&tp->cv_nonfull, &tp->mtx);
    if (tp->stop) { pthread_mutex_unlock(&tp->mtx); free(t); return; }
    if (tp->tail) tp->tail->next = t; else tp->head = t;
    tp->tail = t; tp->size++;
    pthread_cond_signal(&tp->cv_nonempty);
    pthread_mutex_unlock(&tp->mtx);
}

void tp_destroy(threadpool_t *tp) {
    pthread_mutex_lock(&tp->mtx);
    tp->stop = true;
    pthread_cond_broadcast(&tp->cv_nonempty);
    pthread_cond_broadcast(&tp->cv_nonfull);
    pthread_mutex_unlock(&tp->mtx);
    for (int i = 0; i < tp->nthreads; ++i)
        pthread_join(tp->threads[i], NULL);
    free(tp->threads);
    pthread_mutex_destroy(&tp->mtx);
    pthread_cond_destroy(&tp->cv_nonempty);
    pthread_cond_destroy(&tp->cv_nonfull);
}