typestar

thread_pool.c in C

A fixed pool of workers pulling jobs off a shared queue until it closes.

#include <pthread.h>
#include <stdio.h>

#define WORKERS 4
#define QUEUE_CAP 32

typedef struct {
    int jobs[QUEUE_CAP];
    size_t head;
    size_t count;
    int closed;
    long results[QUEUE_CAP];
    pthread_mutex_t lock;
    pthread_cond_t ready;
} Pool;

static void pool_init(Pool *p) {
    p->head = 0;
    p->count = 0;
    p->closed = 0;
    pthread_mutex_init(&p->lock, NULL);
    pthread_cond_init(&p->ready, NULL);
}

static int pool_submit(Pool *p, int job) {
    pthread_mutex_lock(&p->lock);
    int ok = p->count < QUEUE_CAP;
    if (ok) {
        p->jobs[(p->head + p->count) % QUEUE_CAP] = job;
        p->count++;
        pthread_cond_signal(&p->ready);
    }
    pthread_mutex_unlock(&p->lock);
    return ok;
}

static void pool_close(Pool *p) {
    pthread_mutex_lock(&p->lock);
    p->closed = 1;
    pthread_cond_broadcast(&p->ready);
    pthread_mutex_unlock(&p->lock);
}

static void *worker(void *arg) {
    Pool *p = arg;

    for (;;) {
        pthread_mutex_lock(&p->lock);
        while (p->count == 0 && !p->closed) {
            pthread_cond_wait(&p->ready, &p->lock);
        }
        if (p->count == 0 && p->closed) {
            pthread_mutex_unlock(&p->lock);
            return NULL;
        }
        int job = p->jobs[p->head];
        p->head = (p->head + 1) % QUEUE_CAP;
        p->count--;
        pthread_mutex_unlock(&p->lock);

        long squared = (long) job * job;
        pthread_mutex_lock(&p->lock);
        p->results[job % QUEUE_CAP] = squared;
        pthread_mutex_unlock(&p->lock);
    }
}

int main(void) {
    Pool pool;
    pool_init(&pool);

    pthread_t threads[WORKERS];
    for (int i = 0; i < WORKERS; i++) {
        if (pthread_create(&threads[i], NULL, worker, &pool) != 0) {
            fprintf(stderr, "cannot start worker %d\n", i);
            return 1;
        }
    }

    for (int job = 1; job <= 12; job++) {
        if (!pool_submit(&pool, job)) {
            fprintf(stderr, "queue full at %d\n", job);
        }
    }
    pool_close(&pool);

    for (int i = 0; i < WORKERS; i++) {
        pthread_join(threads[i], NULL);
    }

    for (int job = 1; job <= 12; job += 4) {
        printf("%d squared is %ld\n", job, pool.results[job % QUEUE_CAP]);
    }
    return 0;
}

How it works

  1. The queue is guarded by a mutex and a condition variable.
  2. Workers block on the condition rather than spinning.
  3. A shutdown flag plus a broadcast is how the pool ends cleanly.

Keywords and builtins used here

The run, in numbers

Lines
95
Characters to type
1855
Tokens
605
Three-star pace
110 tpm

At the three-star pace of 110 tokens a minute, this run takes about 330 seconds.

Type this snippet

Step 1 of 1 in Encore, step 10 of 10 in Threads & concurrency.

← Previous