Parallel Map with Thread Pools

1. The map Function: A Functional Programming Staple

Before we parallelize anything, let's clarify our goal. The map function is a concept from functional programming. It takes a sequence (like an array) and a function, then applies that function to every element in the sequence, producing a new sequence of the results.

A simple, single-threaded map in C might look like this:

// Applies function 'f' to each element of 'in_array', storing result in 'out_array'.
void sequential_map(void** in_array, void** out_array, size_t len, void* (*f)(void*)) {
    for (size_t i = 0; i < len; ++i) {
        out_array[i] = f(in_array[i]);
    }
}

The challenge and goal for today is to make this operation concurrent. Instead of processing elements one by one, we want to process many of them simultaneously.

The following article provides a great introduction to this concept, showing both a single-threaded and a multi-threaded approach.

The strategy outlined in the article—creating a new thread for each partition—is a valid start. However, the overhead of creating and destroying threads can be significant, especially if we have thousands of small tasks. A much more efficient solution is to use a thread pool.

2. The Thread Pool: A Reusable Engine for Tasks

A thread pool is a group of pre-created worker threads waiting to execute tasks. Instead of spawning a new thread for every task, the main thread submits tasks to a queue. The worker threads consume tasks from this queue, execute them, and then go back to wait for more.

This is the producer-consumer pattern again, but for tasks rather than resources.

To build a generic thread pool, we need a way to represent any kind of task. A common and powerful way to do this is with a struct that holds a function pointer and a void* pointer for its argument.

// Represents a single unit of work for the thread pool
typedef void (*thread_func_t)(void *arg);

typedef struct tpool_work {
    thread_func_t      func; // The function to execute
    void              *arg;  // The argument to the function
    struct tpool_work *next; // Pointer to the next work item
} tpool_work_t;

With this, we can build a full thread pool. The implementation involves a task queue (often a linked list), a mutex to protect the queue, and condition variables to signal between the main thread (producer) and worker threads (consumers).

The following resource provides a complete, well-documented C implementation of a thread pool that we will use as our foundation.

3. Implementing parallel_map with a Thread Pool

Now we can combine these two concepts. Our parallel_map function will use the thread pool from John's Blog to execute a given function on every element of an array.

The strategy is simple:

1. Iterate through the input array.
2. For each element, create a "task" that bundles the function to be applied and a pointer to that element.
3. Add this task to the thread pool's work queue.
4. After adding all tasks, wait for the thread pool to finish processing everything.

Step 1: Defining the Task Arguments

Since our tpool_add_work takes a function and a single void* argument, we need a way to pass both the element's data and the function we want to apply. We can't do this directly.

Instead, we will create a "trampoline" function. The task submitted to the pool will always be this trampoline. The argument to the trampoline will be a struct containing the real function and the real argument (the array element).

// A struct to hold the arguments for our map task
typedef struct {
    void (*map_func)(void *data); // The function to apply (e.g., square_int)
    void *data;                  // A pointer to the array element
} map_task_arg_t;

// This is the "trampoline" function that worker threads will execute.
// It unpacks the arguments and calls the real map function.
void map_trampoline(void *arg) {
    map_task_arg_t *task = (map_task_arg_t *)arg;
    task->map_func(task->data); // Call the user-provided function on the data
    free(task);                 // Clean up the argument struct
}

Notice that map_trampoline is responsible for freeing the map_task_arg_t struct. This is crucial because we will need to dynamically allocate one for each element.

Step 2: The parallel_map Function

Now we can write the main parallel_map function. It orchestrates the process of creating and submitting all the tasks.

/*
 * Applies a function to each element of an array in parallel.
 * This implementation performs an "in-place" map.
 *
 * @param pool The thread pool to use for execution.
 * @param array An array of pointers to the data elements.
 * @param len The number of elements in the array.
 * @param map_func The function to apply to each element.
 */
void parallel_map_inplace(tpool_t *pool, void **array, size_t len, void (*map_func)(void *)) {
    if (pool == NULL || array == NULL || map_func == NULL) {
        return; // Or handle error appropriately
    }

    // 1. Submit a task for each element in the array
    for (size_t i = 0; i < len; ++i) {
        // Allocate task arguments on the heap.
        // The trampoline function will free this memory.
        map_task_arg_t *task_args = malloc(sizeof(map_task_arg_t));
        if (task_args == NULL) {
            // Handle allocation failure
            continue;
        }
        task_args->map_func = map_func;
        task_args->data = array[i];

        // Add the trampoline function with its arguments to the pool
        tpool_add_work(pool, map_trampoline, task_args);
    }

    // 2. Wait for all tasks to complete
    tpool_wait(pool);
}

Step 3: Putting It All Together

Let's see it in action. We'll create a pool, an array of integers, and a function to square them. Then we'll call our parallel_map_inplace and verify the results.

#include <stdio.h>
#include <stdlib.h>
#include "tpool.h" // Assumes the thread pool implementation is in tpool.h/tpool.c

// A simple function to square an integer.
// The argument is a pointer to an int.
void square_int(void *arg) {
    int *val = (int *)arg;
    *val = (*val) * (*val);
    // In a real scenario, you might add a small sleep here
    // to simulate more substantial work.
}

#define NUM_ELEMENTS 100
#define NUM_THREADS 4

int main() {
    // ---- Setup ----
    int *data = malloc(sizeof(int) * NUM_ELEMENTS);
    void **array = malloc(sizeof(void*) * NUM_ELEMENTS);
    for (size_t i = 0; i < NUM_ELEMENTS; ++i) {
        data[i] = i;
        array[i] = &data[i]; // array holds pointers to our data
    }

    printf("Original data:\n");
    for (size_t i = 0; i < 5; ++i) printf("%d ", *(int*)array[i]);
    printf("...\n");

    tpool_t *pool = tpool_create(NUM_THREADS);

    // ---- Execution ----
    parallel_map_inplace(pool, array, NUM_ELEMENTS, square_int);

    // ---- Verification ----
    printf("Mapped data:\n");
    for (size_t i = 0; i < 5; ++i) printf("%d ", *(int*)array[i]);
    printf("...\n");
    
    // Check one of the results
    printf("Element 10 was %d, is now %d\n", 10, *(int*)array[10]);

    // ---- Teardown ----
    tpool_destroy(pool);
    free(data);
    free(array);

    return 0;
}

This example demonstrates the complete workflow. The parallel_map_inplace function is completely generic. It doesn't know it's squaring integers; it only knows how to use the thread pool to apply some function to some data, which makes it a highly reusable and powerful tool.

Conclusion

In this lesson, you leveraged a thread pool to build a generic, parallel map function. This is a significant capstone achievement, bringing together many concepts from this course: function pointers, void* for generic programming, thread-safe queues, mutexes, condition variables, and task-based parallelism.

Key Takeaways:

• A parallel map applies a function to all elements of a collection concurrently, significantly speeding up data processing tasks.
• A thread pool is an efficient engine for executing parallel map tasks, as it avoids the high cost of thread creation and destruction by reusing a fixed set of worker threads.
• Generic task submission can be achieved by defining a work struct containing a function pointer and a void* argument.
• A trampoline function can be used as a wrapper to adapt a more complex function signature (like our map function and its data) to the simple void (*f)(void*) signature expected by the thread pool.
• The main thread must use a barrier-like mechanism (e.g., tpool_wait) to ensure all parallel tasks are complete before proceeding.

Preview of the Next Lesson:
Congratulations on completing this capstone project! You've built a work-stealing queue for dynamic load balancing, a connection pool for managing resources, and now a parallel map function for data processing. These are advanced, practical patterns used in real-world systems.

Moving forward in the course, the next module, "Atomic Operations and Lock-Free Basics," will introduce you to an even lower level of concurrency control. We'll explore how to build certain data structures without using locks at all, relying instead on atomic CPU instructions. Our first topic will be using C11 _Atomic types to implement a lock-free counter.