如何用pthreads并行化KNN代码？排查段错误问题

架构老林

2026-8-18

问题：pthreads实现KNN并行化时出现段错误

我正在完成课程作业，尝试用pthreads实现KNN算法的并行化。已定义结构体并尝试向run2线程函数传递变量，但运行时出现segmentation fault (core dumped)错误。原本的KNN逻辑在KNN_POSIX2函数中，拆分到run2准备多线程实现，先尝试单POSIX线程也失败。刚接触C语言，恳请指教。

代码实现

typedef struct PASSING_PARAMS2 {
    int* thread_identifier;
    ArffData* traindata;
    ArffData* testdata;
    int k;
    int tcount;
    int qIndex;
    int* parampred;
    int* paramclasscount;
    float* paramcandidates;
    int paramnumclasses;
} PassingParams2;

void* run2(void* ptr) {
    PassingParams2 *paramPtr = (PassingParams2 *)ptr;
    ArffData* train = (ArffData*)paramPtr->traindata;
    ArffData* test = (ArffData*)paramPtr->testdata;
    float* candidates = (float*)paramPtr->paramcandidates;
    int k = (int)paramPtr->k;
    int* classCounts = (int*)paramPtr->paramclasscount;
    int num_classes = (int) paramPtr->paramnumclasses;

    for(int queryIndex = 0; queryIndex < test->num_instances(); queryIndex++) {
        for(int keyIndex = 0; keyIndex < train->num_instances(); keyIndex++) {
            
            float dist = distance(test->get_instance(queryIndex), train->get_instance(keyIndex));

            // Add to our candidates
            for(int c = 0; c < k; c++){
                if(dist < candidates[2*c]){
                    // Found a new candidate
                    // Shift previous candidates down by one
                    for(int x = k-2; x >= c; x--) {
                        candidates[2*x+2] = candidates[2*x];
                        candidates[2*x+3] = candidates[2*x+1];
                    }
                    
                    // Set key vector as potential k NN
                    candidates[2*c] = dist;
                    candidates[2*c+1] = train->get_instance(keyIndex)->get(train->num_attributes() - 1)->operator float(); // class value

                    break;
                }
            }
        }

        // Bincount the candidate labels and pick the most common
        for(int i = 0; i < k;i++){
            classCounts[(int)candidates[2*i+1]] += 1;
        }
        
        int max = -1;
        int max_index = 0;
        for(int i = 0; i < num_classes;i++){
            if(classCounts[i] > max){
                max = classCounts[i];
                max_index = i;
            }
        }

        predictions[queryIndex] = max_index;

        for(int i = 0; i < 2*k; i++){ candidates[i] = FLT_MAX; }
        memset(classCounts, 0, num_classes * sizeof(int));
    }
    pthread_exit(0);
}

int* KNN_POSIX2(ArffData* train, ArffData* test, int k, int t) {

    // Predictions is the array where you have to return the class predicted (integer) for the test dataset instances
    int* predictions = (int*)malloc(test->num_instances() * sizeof(int));

    // Stores k-NN candidates for a query vector as a sorted 2d array. First element is inner product, second is class.
    float* candidates = (float*) calloc(k*2, sizeof(float));
    for(int i = 0; i < 2*k; i++){ candidates[i] = FLT_MAX; }

    int num_classes = train->num_classes();

    // Stores bincounts of each class over the final set of candidate NN
    int* classCounts = (int*)calloc(num_classes, sizeof(int));
    
    //Setup of the parameters to be passed by the struct
    PassingParams2 *paramPtr;
    paramPtr = (PassingParams2 *) malloc(1* sizeof(PassingParams2));
    paramPtr->thread_identifier = (int*)malloc(1 * sizeof(int));
    paramPtr->traindata = train;
    paramPtr->k = k;
    paramPtr->tcount = t;
    paramPtr->paramnumclasses = num_classes;

    // create the threads
    pthread_t newthread;
    pthread_create(&newthread, NULL, &run2, (void*) paramPtr);

    return predictions;
}

错误原因及修复方案

1. 结构体参数未完全初始化

KNN_POSIX2中只给结构体的部分字段赋值，testdata、paramcandidates、paramclasscount、parampred这些字段都是空指针，线程函数run2访问这些空指针直接触发段错误。

修复：在结构体初始化时补全所有字段：

// 补全结构体字段赋值
paramPtr->testdata = test;
paramPtr->paramcandidates = candidates;
paramPtr->paramclasscount = classCounts;
paramPtr->parampred = predictions;

2. 未等待线程执行完成就返回

KNN_POSIX2创建线程后直接返回predictions，主线程可能提前结束，导致线程还在运行时进程终止，或者predictions等内存虽然是堆分配，但线程未完成写入就返回，结果不可靠。

修复：调用pthread_join等待线程执行完毕：

// 创建线程后添加等待逻辑
pthread_create(&newthread, NULL, &run2, (void*) paramPtr);
pthread_join(newthread, NULL); // 等待线程完成

3. 线程函数访问未定义的`predictions`变量

run2中的predictions[queryIndex]是未定义的全局变量（或主线程局部变量，线程无法直接访问），必须通过结构体传递的parampred字段访问。

修复：修改run2中的赋值语句：

// 替换原predictions[queryIndex] = max_index;
paramPtr->parampred[queryIndex] = max_index;

4. 内存泄漏与冗余分配

结构体中的thread_identifier字段未使用，且分配了内存未释放，属于冗余。可以直接删除该字段，或者使用后释放。另外，结构体本身在使用后也需要释放（线程函数中可以释放结构体指针）。

修复：

若不需要thread_identifier，直接从结构体定义中删除该字段，同时去掉对应的malloc代码。
在run2函数末尾释放结构体指针（如果结构体是动态分配的）：

free(paramPtr);
pthread_exit(0);

5. 不必要的类型转换

结构体中的字段已经是对应类型，无需额外强制转换，简化代码：

// 原代码中的强制转换可以去掉
ArffData* train = paramPtr->traindata;
ArffData* test = paramPtr->testdata;
float* candidates = paramPtr->paramcandidates;
int k = paramPtr->k;
int* classCounts = paramPtr->paramclasscount;
int num_classes = paramPtr->paramnumclasses;

修复后的完整代码示例

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

// 假设ArffData及相关方法已定义
typedef struct ArffData ArffData;

typedef struct PASSING_PARAMS2 {
    ArffData* traindata;
    ArffData* testdata;
    int k;
    int tcount;
    int qIndex;
    int* parampred;
    int* paramclasscount;
    float* paramcandidates;
    int paramnumclasses;
} PassingParams2;

float distance(void* test_inst, void* train_inst) {
    // 假设距离计算逻辑已实现
    return 0.0f;
}

void* run2(void* ptr) {
    PassingParams2 *paramPtr = (PassingParams2 *)ptr;
    ArffData* train = paramPtr->traindata;
    ArffData* test = paramPtr->testdata;
    float* candidates = paramPtr->paramcandidates;
    int k = paramPtr->k;
    int* classCounts = paramPtr->paramclasscount;
    int num_classes = paramPtr->paramnumclasses;
    int* predictions = paramPtr->parampred;

    for(int queryIndex = 0; queryIndex < test->num_instances(); queryIndex++) {
        for(int keyIndex = 0; keyIndex < train->num_instances(); keyIndex++) {
            
            float dist = distance(test->get_instance(queryIndex), train->get_instance(keyIndex));

            // Add to our candidates
            for(int c = 0; c < k; c++){
                if(dist < candidates[2*c]){
                    // Found a new candidate
                    // Shift previous candidates down by one
                    for(int x = k-2; x >= c; x--) {
                        candidates[2*x+2] = candidates[2*x];
                        candidates[2*x+3] = candidates[2*x+1];
                    }
                    
                    // Set key vector as potential k NN
                    candidates[2*c] = dist;
                    candidates[2*c+1] = train->get_instance(keyIndex)->get(train->num_attributes() - 1)->operator float(); // class value

                    break;
                }
            }
        }

        // Bincount the candidate labels and pick the most common
        for(int i = 0; i < k;i++){
            classCounts[(int)candidates[2*i+1]] += 1;
        }
        
        int max = -1;
        int max_index = 0;
        for(int i = 0; i < num_classes;i++){
            if(classCounts[i] > max){
                max = classCounts[i];
                max_index = i;
            }
        }

        predictions[queryIndex] = max_index;

        for(int i = 0; i < 2*k; i++){ candidates[i] = FLT_MAX; }
        memset(classCounts, 0, num_classes * sizeof(int));
    }

    free(paramPtr);
    pthread_exit(NULL);
}

int* KNN_POSIX2(ArffData* train, ArffData* test, int k, int t) {

    int* predictions = (int*)malloc(test->num_instances() * sizeof(int));
    if (!predictions) return NULL;

    float* candidates = (float*) calloc(k*2, sizeof(float));
    if (!candidates) {
        free(predictions);
        return NULL;
    }
    for(int i = 0; i < 2*k; i++){ candidates[i] = FLT_MAX; }

    int num_classes = train->num_classes();

    int* classCounts = (int*)calloc(num_classes, sizeof(int));
    if (!classCounts) {
        free(candidates);
        free(predictions);
        return NULL;
    }
    
    PassingParams2 *paramPtr = (PassingParams2 *) malloc(sizeof(PassingParams2));
    if (!paramPtr) {
        free(classCounts);
        free(candidates);
        free(predictions);
        return NULL;
    }

    paramPtr->traindata = train;
    paramPtr->testdata = test;
    paramPtr->k = k;
    paramPtr->tcount = t;
    paramPtr->paramnumclasses = num_classes;
    paramPtr->parampred = predictions;
    paramPtr->paramclasscount = classCounts;
    paramPtr->paramcandidates = candidates;

    pthread_t newthread;
    int ret = pthread_create(&newthread, NULL, run2, (void*) paramPtr);
    if (ret != 0) {
        free(paramPtr);
        free(classCounts);
        free(candidates);
        free(predictions);
        return NULL;
    }

    pthread_join(newthread, NULL);

    // 释放线程中使用的临时内存
    free(classCounts);
    free(candidates);

    return predictions;
}

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

如何用pthreads并行化KNN代码？排查段错误问题

代码实现 #

错误原因及修复方案 #

1. 结构体参数未完全初始化 #

2. 未等待线程执行完成就返回 #

3. 线程函数访问未定义的predictions变量 #

4. 内存泄漏与冗余分配 #

5. 不必要的类型转换 #

修复后的完整代码示例 #