typestar

thread_pool.c en C

Un pool fijo de workers sacando trabajos de una cola compartida hasta que cierra.

#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;
}

Cómo funciona

  1. La cola está protegida por un mutex y una variable de condición.
  2. Los workers bloquean en la condición en vez de girar en vacío.
  3. Un flag de apagado más un broadcast es como el pool termina limpio.

Palabras clave y builtins usados aquí

El intento, en números

Líneas
95
Caracteres a escribir
1855
Tokens
605
Ritmo de tres estrellas
110 tpm

Al ritmo de tres estrellas de 110 tokens por minuto, este intento toma unos 330 segundos.

Escribe este fragmento

Paso 1 de 1 en Bis; paso 10 de 10 en Hilos y concurrencia.

← Anterior