23. Mini projects: a thread pool, and a select()-based chat server

23.1 Mini project — a thread pool

The pattern behind every real worker-pool (web server request handlers, Section 26's shell running background jobs, Redis's background I/O threads): a fixed number of worker threads pull tasks off a shared queue protected by a mutex + condition variable.

// threadpool.c
#include <pthread.h>
#include <stdio.h>
#include <stdlib.h>

#define NUM_THREADS 4
#define QUEUE_CAP   64

typedef void (*Task)(void *);

typedef struct { Task fn; void *arg; } Job;

static Job      queue[QUEUE_CAP];
static int      q_head = 0, q_tail = 0, q_count = 0;
static pthread_mutex_t q_lock = PTHREAD_MUTEX_INITIALIZER;
static pthread_cond_t  q_not_empty = PTHREAD_COND_INITIALIZER;
static int      shutdown_flag = 0;

void pool_submit(Task fn, void *arg) {
    pthread_mutex_lock(&q_lock);
    queue[q_tail] = (Job){ fn, arg };
    q_tail = (q_tail + 1) % QUEUE_CAP;
    q_count++;
    pthread_cond_signal(&q_not_empty);
    pthread_mutex_unlock(&q_lock);
}

void *worker(void *arg) {
    int id = *(int *)arg;
    while (1) {
        pthread_mutex_lock(&q_lock);
        while (q_count == 0 && !shutdown_flag) pthread_cond_wait(&q_not_empty, &q_lock);
        if (q_count == 0 && shutdown_flag) { pthread_mutex_unlock(&q_lock); break; }

        Job job = queue[q_head];
        q_head = (q_head + 1) % QUEUE_CAP;
        q_count--;
        pthread_mutex_unlock(&q_lock);

        job.fn(job.arg);   // run the task OUTSIDE the lock — never hold a lock during work
    }
    printf("worker %d exiting\n", id);
    return NULL;
}

void print_job(void *arg) {
    printf("job %d running on thread %lu\n", *(int *)arg, (unsigned long)pthread_self());
    free(arg);
}

int main(void) {
    pthread_t threads[NUM_THREADS];
    int ids[NUM_THREADS];
    for (int i = 0; i < NUM_THREADS; i++) {
        ids[i] = i;
        pthread_create(&threads[i], NULL, worker, &ids[i]);
    }

    for (int i = 0; i < 20; i++) {
        int *job_id = malloc(sizeof(int));
        if (!job_id) { perror("malloc"); break; }
        *job_id = i;
        pool_submit(print_job, job_id);
    }

    pthread_mutex_lock(&q_lock);
    shutdown_flag = 1;
    pthread_cond_broadcast(&q_not_empty);   // wake ALL workers so they can see shutdown_flag
    pthread_mutex_unlock(&q_lock);

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

23.2 Mini project — a select()-based multi-client chat server

This fixes Section 21.1's "only one client at a time" limitation without threads at all — a single-threaded event loop, the same core idea behind libuv (Node.js), libevent, and Redis's ae.c (Section 34).

// chat_server.c
#include <stdio.h>
#include <string.h>
#include <unistd.h>
#include <arpa/inet.h>
#include <sys/socket.h>
#include <sys/select.h>

#define MAX_CLIENTS 32

int main(void) {
    int server_fd = socket(AF_INET, SOCK_STREAM, 0);
    int opt = 1;
    setsockopt(server_fd, SOL_SOCKET, SO_REUSEADDR, &opt, sizeof opt);

    struct sockaddr_in addr = { .sin_family = AF_INET, .sin_port = htons(9000),
                                 .sin_addr.s_addr = INADDR_ANY };
    bind(server_fd, (struct sockaddr *)&addr, sizeof addr);
    listen(server_fd, 10);

    int clients[MAX_CLIENTS] = {0};   // 0 means "empty slot"
    printf("chat server on :9000\n");

    while (1) {
        fd_set readfds;
        FD_ZERO(&readfds);
        FD_SET(server_fd, &readfds);
        int max_fd = server_fd;

        for (int i = 0; i < MAX_CLIENTS; i++) {
            if (clients[i] > 0) {
                FD_SET(clients[i], &readfds);
                if (clients[i] > max_fd) max_fd = clients[i];
            }
        }

        select(max_fd + 1, &readfds, NULL, NULL, NULL);   // block until ANY fd is ready

        if (FD_ISSET(server_fd, &readfds)) {               // new connection waiting
            int new_fd = accept(server_fd, NULL, NULL);
            for (int i = 0; i < MAX_CLIENTS; i++) {
                if (clients[i] == 0) { clients[i] = new_fd; break; }
            }
        }

        for (int i = 0; i < MAX_CLIENTS; i++) {
            if (clients[i] > 0 && FD_ISSET(clients[i], &readfds)) {
                char buf[512];
                ssize_t n = read(clients[i], buf, sizeof buf);
                if (n <= 0) {                               // client disconnected
                    close(clients[i]);
                    clients[i] = 0;
                    continue;
                }
                for (int j = 0; j < MAX_CLIENTS; j++) {      // broadcast to everyone else
                    if (clients[j] > 0 && clients[j] != clients[i]) {
                        write(clients[j], buf, (size_t)n);
                    }
                }
            }
        }
    }
}

select() scales poorly past a few hundred descriptors (it's O(n) per call and has a hard FD_SETSIZE limit) — production event loops use epoll (Linux) or kqueue (BSD/macOS) instead, which is exactly what Redis's ae.c abstracts over in Section 34.