Worker Pools in C

Our example demonstrates how to implement a worker pool using threads and shared memory in C.

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

#define NUM_WORKERS 3
#define NUM_JOBS 5

// Shared data structure for jobs and results
struct shared_data {
    int jobs[NUM_JOBS];
    int results[NUM_JOBS];
    int job_index;
    int result_index;
    pthread_mutex_t mutex;
    pthread_cond_t job_available;
    pthread_cond_t result_available;
};

// Worker function
void* worker(void* arg) {
    struct shared_data* data = (struct shared_data*)arg;
    int id = rand() % 1000;  // Generate a random worker ID

    while (1) {
        int job;
        pthread_mutex_lock(&data->mutex);

        while (data->job_index >= NUM_JOBS) {
            pthread_cond_wait(&data->job_available, &data->mutex);
            if (data->job_index >= NUM_JOBS) {
                pthread_mutex_unlock(&data->mutex);
                return NULL;  // No more jobs, exit the thread
            }
        }

        job = data->jobs[data->job_index++];
        pthread_mutex_unlock(&data->mutex);

        printf("worker %d started  job %d\n", id, job);
        sleep(1);  // Simulate work
        printf("worker %d finished job %d\n", id, job);

        pthread_mutex_lock(&data->mutex);
        data->results[data->result_index++] = job * 2;
        pthread_cond_signal(&data->result_available);
        pthread_mutex_unlock(&data->mutex);
    }
}

int main() {
    struct shared_data data = {
        .job_index = 0,
        .result_index = 0
    };
    pthread_t workers[NUM_WORKERS];

    pthread_mutex_init(&data.mutex, NULL);
    pthread_cond_init(&data.job_available, NULL);
    pthread_cond_init(&data.result_available, NULL);

    // Create worker threads
    for (int w = 0; w < NUM_WORKERS; w++) {
        pthread_create(&workers[w], NULL, worker, &data);
    }

    // Send jobs
    for (int j = 1; j <= NUM_JOBS; j++) {
        pthread_mutex_lock(&data.mutex);
        data.jobs[data.job_index++] = j;
        pthread_cond_signal(&data.job_available);
        pthread_mutex_unlock(&data.mutex);
    }

    // Wait for all results
    for (int a = 0; a < NUM_JOBS; a++) {
        pthread_mutex_lock(&data.mutex);
        while (data.result_index <= a) {
            pthread_cond_wait(&data.result_available, &data.mutex);
        }
        pthread_mutex_unlock(&data.mutex);
    }

    // Signal workers to exit and wait for them
    pthread_mutex_lock(&data.mutex);
    data.job_index = NUM_JOBS;
    pthread_cond_broadcast(&data.job_available);
    pthread_mutex_unlock(&data.mutex);

    for (int w = 0; w < NUM_WORKERS; w++) {
        pthread_join(workers[w], NULL);
    }

    pthread_mutex_destroy(&data.mutex);
    pthread_cond_destroy(&data.job_available);
    pthread_cond_destroy(&data.result_available);

    return 0;
}

This C program implements a worker pool using POSIX threads (pthreads). Here’s a breakdown of how it works:

We define constants for the number of workers and jobs.
A shared_data structure is used to manage jobs, results, and synchronization primitives.
The worker function represents each worker thread. It waits for available jobs, processes them, and stores the results.
In the main function, we:
- Initialize the shared data and create worker threads.
- Send jobs to the workers.
- Wait for all results to be processed.
- Signal the workers to exit and wait for them to finish.
We use mutexes and condition variables for thread synchronization:
- mutex ensures exclusive access to shared data.
- job_available signals when new jobs are available.
- result_available signals when results are ready.

To compile and run this program:

$ gcc -o worker_pool worker_pool.c -lpthread
$ ./worker_pool
worker 383 started  job 1
worker 886 started  job 2
worker 777 started  job 3
worker 383 finished job 1
worker 383 started  job 4
worker 886 finished job 2
worker 886 started  job 5
worker 777 finished job 3
worker 383 finished job 4
worker 886 finished job 5

The output shows the 5 jobs being executed by various workers. The program takes about 2 seconds to complete, despite doing about 5 seconds of total work, because there are 3 workers operating concurrently.

Note that unlike goroutines, which are lightweight and managed by the runtime, this C implementation uses actual OS threads, which are more resource-intensive. For large-scale concurrent operations, you might need to consider more advanced techniques or libraries designed for high-concurrency scenarios in C.

Tickers in C

Waitgroups in C

Learn X By Example

Title here

Worker Pools in C