Multi-Threaded Event Processing System

1. The Architectural Blueprint: The Boss/Worker Pattern

An event processing system needs to be responsive. It cannot get bogged down handling a long-running task while new events are arriving. The key to achieving this is a division of labor. The most common and effective way to structure this is using the Boss/Worker design pattern.

As the article explains, the architecture consists of three main components:

1. The Dispatcher (Boss) Thread: A single thread whose only job is to wait for incoming events and place them into a shared task queue. In a network server, this "event" would be a new client connection returned by accept(). By keeping the dispatcher's work minimal, it remains highly responsive and can quickly handle a high rate of incoming events.
2. The Task Queue: A thread-safe data structure (like the queues we've built in previous lessons) that holds the events waiting to be processed. This queue acts as a buffer, smoothing out bursts of incoming events.
3. The Worker Threads: A pool of pre-created threads that wait for tasks to appear in the queue. When a task is available, a worker retrieves it, processes it completely, and then returns to the queue to wait for another. This is where the actual "work" gets done.

This model is visualized perfectly by the thread pool architecture we've seen before.

A GitHub project for an event loop in C provides a good summary of how these components fit together in a server context.

2. Implementing the Engine: A Thread Pool with Efficient Waiting

We know from previous lessons that creating a new thread for every event is inefficient due to the overhead of thread creation and destruction. A thread pool is the solution. However, implementing the worker logic requires care to avoid being inefficient in a different way.

Let's watch a video that walks through building a multithreaded server, starting with a simple (but flawed) approach and refining it.

The busy-waiting problem is exactly what condition variables are designed to solve. Instead of spinning, a worker thread should sleep efficiently when the queue is empty and be woken up only when the dispatcher adds a new event.

This brings us to the canonical implementation of a worker thread's main loop, which you've implemented before: the wait-in-a-loop pattern.

3. A Complete Implementation

Let's tie this all together into a working C program. We will simulate an event processing system.

• Our "events" will be simple structs containing an ID.
• The dispatcher will be the main function, generating events in a loop.
• The workers will process events by printing the event ID and sleeping for a short time to simulate work.
• We'll use a simple, fixed-size array as our task queue, protected by a mutex and a condition variable.

Here is the complete code:

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

#define NUM_WORKERS 4
#define QUEUE_CAPACITY 16

// Our simple "event" or "task"
typedef struct {
    int id;
} event_t;

// The shared task queue
event_t task_queue[QUEUE_CAPACITY];
int task_count = 0;
int queue_head = 0;
int queue_tail = 0;

// Synchronization primitives
pthread_mutex_t queue_mutex;
pthread_cond_t queue_cond;

// Function for the worker threads
void* worker_function(void* arg) {
    long worker_id = (long)arg;
    printf("Worker %ld started.\n", worker_id);

    while (1) {
        pthread_mutex_lock(&queue_mutex);

        // Wait-in-a-loop pattern to handle spurious wakeups
        while (task_count == 0) {
            printf("Worker %ld: Queue empty, waiting...\n", worker_id);
            pthread_cond_wait(&queue_cond, &queue_mutex);
            printf("Worker %ld: Woke up.\n", worker_id);
        }

        // Dequeue an event
        event_t event = task_queue[queue_head];
        queue_head = (queue_head + 1) % QUEUE_CAPACITY;
        task_count--;

        pthread_mutex_unlock(&queue_mutex);

        // Process the event (outside the lock)
        printf("Worker %ld: Processing event %d.\n", worker_id, event.id);
        sleep(1); // Simulate work
    }

    return NULL;
}

// Function for the dispatcher thread (main)
int main() {
    pthread_t workers[NUM_WORKERS];

    // Initialize mutex and condition variable
    pthread_mutex_init(&queue_mutex, NULL);
    pthread_cond_init(&queue_cond, NULL);

    // Create worker threads
    for (long i = 0; i < NUM_WORKERS; ++i) {
        pthread_create(&workers[i], NULL, worker_function, (void*)i);
    }

    // --- This is the dispatcher loop ---
    int event_id = 0;
    while (1) {
        // "Generate" a new event
        event_t new_event = { .id = event_id++ };
        
        usleep(500 * 1000); // Dispatcher generates events every 500ms

        pthread_mutex_lock(&queue_mutex);

        // Enqueue the event
        if (task_count < QUEUE_CAPACITY) {
            task_queue[queue_tail] = new_event;
            queue_tail = (queue_tail + 1) % QUEUE_CAPACITY;
            task_count++;
            printf("Dispatcher: Enqueued event %d. (Queue size: %d)\n", new_event.id, task_count);

            // Signal ONE waiting worker thread
            pthread_cond_signal(&queue_cond);
        } else {
            printf("Dispatcher: Queue full, dropping event %d.\n", new_event.id);
        }

        pthread_mutex_unlock(&queue_mutex);
    }

    // In a real application, you would have a graceful shutdown mechanism
    // to join the worker threads. For this example, we'll let it run forever.
    // for (int i = 0; i < NUM_WORKERS; ++i) {
    //     pthread_join(workers[i], NULL);
    // }
    // pthread_mutex_destroy(&queue_mutex);
    // pthread_cond_destroy(&queue_cond);

    return 0;
}

Compile this code with gcc -o event_system event_system.c -pthread and run it. You will see the dispatcher steadily adding events to the queue and the worker threads waking up to process them. You'll see workers waiting when the queue is empty and the dispatcher warning when the queue is full. This is the Boss/Worker pattern in action.

Conclusion

Congratulations on completing the course! In this final lesson, you designed and implemented a multi-threaded event processing system, one of the most fundamental and useful patterns in concurrent C programming.

Key Takeaways:

• The Boss/Worker (Dispatcher/Worker) pattern is a highly effective way to structure concurrent, event-driven applications. It decouples event reception from event processing, improving responsiveness and scalability.
• The Dispatcher's role is minimal: accept an event and enqueue it quickly.
• The Worker's role is to take a task from the queue and process it fully.
• A thread-safe task queue using a mutex for mutual exclusion and a condition variable to avoid busy-waiting is the efficient engine at the heart of this pattern.

You have now built a strong, practical foundation in C multithreading. From the basic building blocks of pthread_create and pthread_mutex_lock to advanced, reusable patterns like thread pools, connection pools, and event-driven systems, you are now equipped to design, implement, and reason about complex concurrent software. The journey from here involves applying these patterns to new problems and continuing to hone your skills in debugging and performance optimization. Well done