如何在C语言中实现结果有序的函数式parallel-map

函数式映射（functional map）的作用是将回调函数应用到数组的每个元素上，返回由回调结果组成的新数组。比如伪代码map(["hello", "world"], fn(x) => x + " meow")会返回["hello meow", "world meow"]。

原有的串行实现如下：

#include <stdio.h>
#include <stdlib.h>
#include <string.h>

// 原串行函数式映射
void** fp_map(void** array, size_t len, void* (*execute)(void*))
{
    // 为结果分配内存
    void** returns = malloc(sizeof(void*) * len);
    if (returns == NULL) {
        fprintf(stderr, "Malloc failed, buy more ram\n");
        exit(42);
    }

    // 逐个映射元素
    for (size_t i = 0; i < len; ++i)
        returns[i] = execute(array[i]);

    return returns;
}

要实现并行映射来提升速度，我们可以借助POSIX线程（pthread）让每个execute()调用在独立线程中执行，同时保证结果数组的顺序和输入数组一致。具体实现如下：

核心实现思路 #

1. 提前分配结果数组，确保每个线程能把结果写入对应索引的位置
2. 定义线程参数结构体，传递每个线程需要的输入元素、回调函数、结果数组以及元素索引
3. 为每个元素创建独立线程，线程执行回调后将结果写入结果数组的对应位置
4. 等待所有线程执行完毕后，返回结果数组

完整并行映射代码 #

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

// 线程参数结构体：打包每个线程所需的所有数据
typedef struct {
    void* element;          // 当前要处理的数组元素
    void* (*execute)(void*);// 回调函数
    void** result_array;    // 结果数组
    size_t index;           // 当前元素在数组中的索引
} ThreadArgs;

// 线程执行函数：调用回调并将结果写入对应位置
void* thread_execute(void* args) {
    ThreadArgs* thread_args = (ThreadArgs*)args;
    thread_args->result_array[thread_args->index] = thread_args->execute(thread_args->element);
    free(args); // 释放参数内存
    return NULL;
}

// 并行函数式映射实现
void** parallel_fp_map(void** array, size_t len, void* (*execute)(void*))
{
    // 提前分配结果数组
    void** returns = malloc(sizeof(void*) * len);
    if (returns == NULL) {
        fprintf(stderr, "Malloc failed, buy more ram\n");
        exit(42);
    }

    // 创建线程数组存储每个线程ID
    pthread_t* threads = malloc(sizeof(pthread_t) * len);
    if (threads == NULL) {
        fprintf(stderr, "Failed to allocate thread IDs\n");
        free(returns);
        exit(43);
    }

    // 为每个元素创建线程
    for (size_t i = 0; i < len; ++i) {
        ThreadArgs* args = malloc(sizeof(ThreadArgs));
        if (args == NULL) {
            fprintf(stderr, "Failed to allocate thread args\n");
            // 清理已创建的线程和内存
            for (size_t j = 0; j < i; ++j)
                pthread_join(threads[j], NULL);
            free(threads);
            free(returns);
            exit(44);
        }
        args->element = array[i];
        args->execute = execute;
        args->result_array = returns;
        args->index = i;

        if (pthread_create(&threads[i], NULL, thread_execute, args) != 0) {
            fprintf(stderr, "Failed to create thread %zu\n", i);
            free(args);
            for (size_t j = 0; j < i; ++j)
                pthread_join(threads[j], NULL);
            free(threads);
            free(returns);
            exit(45);
        }
    }

    // 等待所有线程执行完成
    for (size_t i = 0; i < len; ++i) {
        pthread_join(threads[i], NULL);
    }

    // 清理线程ID数组
    free(threads);
    return returns;
}

// 测试主函数
int main() {
    char* arr[] = {"hello", "world"};
    size_t arr_len = sizeof(arr) / sizeof(arr[0]);

    // 回调函数：给字符串追加" meow"
    void* append_meow(void* x) {
        char* str = (char*)x;
        size_t new_len = strlen(str) + 7; // " meow"含结束符共7个字符
        char* buf = malloc(sizeof(char) * new_len);
        if (buf == NULL) {
            fprintf(stderr, "Malloc failed in callback\n");
            exit(46);
        }
        strcpy(buf, str);
        strcat(buf, " meow");
        return buf;
    }

    // 调用并行映射
    char** arr2 = (char**)parallel_fp_map((void**)arr, arr_len, append_meow);

    // 打印结果并清理内存
    for (size_t i = 0; i < arr_len; ++i) {
        printf("%s\n", arr2[i]);
        free(arr2[i]); // 释放回调中分配的内存
    }
    free(arr2); // 释放结果数组

    return 0;
}

关键细节说明 #

• 顺序保证：通过传递元素索引，每个线程将结果写入结果数组的对应位置，最终结果顺序与输入数组完全一致
• 线程安全：每个线程仅操作结果数组中自己对应的索引位置，无共享写冲突；题目明确回调是纯函数，本身线程安全，无需额外互斥锁
• 错误处理：内存分配或线程创建失败时，会清理已分配资源再退出，避免内存泄漏
• 资源清理：线程参数在线程执行后释放，主函数需手动释放回调分配的内存和结果数组

内容的提问来源于stack exchange，提问作者Hykilpikonna