Python C扩展自定义Pthreads线程池罕见异常问题排查求助

我正在开发一款图计算的Python C扩展，其中一个计算特定图类边列表的例程适合并行化，于是改用线程池实现。最初用OpenMP效果还行，但OpenMP线程池无法在多次Python调用间持久化，所以基于pthreads自定义了线程池。

目前实现几乎能正常运行，但存在两个问题：

• 负载分配不稳定：绝大多数时候所有任务都由单个线程执行，但偶尔会公平分配任务，不确定是否需要调整调度器来稳定负载。
• 偶发段错误：错误出现在多线程（MT）代码执行后的单线程（ST）阶段。MT代码负责计算数组偏移量供ST代码使用，约1%的情况下偏移量内存未完全初始化，导致ST代码越界访问。单元测试对比Python计算结果，99%能通过，说明不是未初始化内存恰好正确的情况；有时ST代码不越界，但会创建超大numpy数组，导致解释器调用耗时极久。

通过printf调试发现，段错误时偏移量内存确实未完全初始化，部分偏移量是百万级垃圾值；99%的情况内存初始化正常。MT代码只读取numpy数组内存，未执行Python代码，按Python C扩展文档这应该没问题。单线程或OpenMP时无此问题，怀疑是自定义线程池的实现问题。调度行为也令人困惑：要么始终公平分配，要么始终单线程执行才合理，但实际时而变化，且该现象和错误无关——错误在两种负载分配情况中都出现过。

核心代码（任务执行与线程同步） #

typedef struct{
    volatile uint capacity;
    volatile void** volatile data;

    volatile uint cursor;
    pthread_mutex_t mutex;
    pthread_cond_t empty_condition;
} Queue;

typedef struct{
    Queue* job_queue;
    void (*job_process)(void*, int ID);

    pthread_mutex_t* volatile barrier_mutex;

    pthread_mutex_t progress_mutex;
    pthread_cond_t progress_condition;
    volatile uint work_in_progress;

    volatile bool terminate;
} SynchronizationHandle;

/* after locking the queue, thread either:
    - executes job from queue asynchronously
    -- this also involves updating and signaling to main thread when job execution is in progress or finished
    - or signals to main thread that queue is empty*/

static bool WorkerThread_do_job(SynchronizationHandle* sync_handle, int ID){  
    Queue* job_queue = sync_handle->job_queue;
    pthread_mutex_lock(&(job_queue->mutex));

    if(job_queue->cursor > 0){
        // grab data from queue and give it free for other threads
        void* job_data = (void*)job_queue->data[job_queue->cursor-1];
        job_queue->cursor--;
        pthread_mutex_unlock(&(job_queue->mutex));

        pthread_mutex_lock(&(sync_handle->progress_mutex));
        sync_handle->work_in_progress++;
        pthread_mutex_unlock(&(sync_handle->progress_mutex));

        sync_handle->job_process(job_data, ID);

        pthread_mutex_lock(&(sync_handle->progress_mutex));
        sync_handle->work_in_progress--;
        if(sync_handle->work_in_progress == 0)
            pthread_cond_signal(&(sync_handle->progress_condition));
        pthread_mutex_unlock(&(sync_handle->progress_mutex));

        return true;
    }
    else{
        pthread_cond_signal(&(job_queue->empty_condition));
        pthread_mutex_unlock(&(job_queue->mutex));
    
        return false;
    }
}

// function for worker threads
// the 2 nested loops correspond to a waiting phase and a work phase
static void* worker_thread(void* arguments){
    WorkerThread* worker_thread = (WorkerThread*) arguments;
    SynchronizationHandle* sync_handle = worker_thread->sync_handle;

    // the outer loop deals with waiting on the main thread to start work or terminate
    while(true){
        // adress is saved to local variable so that main thread can change adress to other mutex without race condition
        pthread_mutex_t* const barrier_mutex = (pthread_mutex_t* const)sync_handle->barrier_mutex;
        // try to lock mutex and go to sleep and wait for main thread to unlock it
        pthread_mutex_lock(barrier_mutex);
        // unlock it for other threads
        pthread_mutex_unlock(barrier_mutex);

        // the inner loop executes jobs from the work queue and checks inbetween if it should terminate
        while(true){
            if(sync_handle->terminate)
                pthread_exit(0);
            if(!WorkerThread_do_job(sync_handle, worker_thread->ID))
                break;
        }
    }
}

typedef struct{
    pthread_t* thread_handles;
    WorkerThread* worker_threads;
    uint worker_threads_count;

    SynchronizationHandle* sync_handle;
    pthread_mutex_t barrier_mutexes[2];
    uint barrier_mutex_cursor;
} ThreadPool;

static void ThreadPool_wakeup_workers(ThreadPool* pool){
    // ASSUMPTION: thread executing already owns pool->barrier_mutexes + pool->barrier_mutex_cursor

    // compute which mutex to switch to next
    uint offset = (pool->barrier_mutex_cursor + 1)%2;

    // lock next mutex before wake up so that worker threads can't get hold of it before main thread
    pthread_mutex_lock(pool->barrier_mutexes + offset);

    // change adress to the next mutex before unlocking previous mutex, otherwise race condition
    pool->sync_handle->barrier_mutex = pool->barrier_mutexes + offset;

    // unlocking the previous mutex "wakes up" the worker threads since they are trying to lock it
    // hence why the assumption needs to hold
    pthread_mutex_unlock(pool->barrier_mutexes + pool->barrier_mutex_cursor);
    
    pool->barrier_mutex_cursor = offset;
}

static void ThreadPool_participate(ThreadPool* pool){
    while(WorkerThread_do_job(pool->sync_handle, 0));
}

static void ThreadPool_waiton_workers(ThreadPool* pool){
    // wait until queue is empty
    pthread_mutex_lock(&(pool->sync_handle->job_queue->mutex));
    while(pool->sync_handle->job_queue->cursor > 0)
        pthread_cond_wait(&(pool->sync_handle->job_queue->empty_condition), &(pool->sync_handle->job_queue->mutex));
    pthread_mutex_unlock(&(pool->sync_handle->job_queue->mutex));

    // wait until all work in progress is finished
    pthread_mutex_lock(&(pool->sync_handle->progress_mutex));
    while(pool->sync_handle->work_in_progress > 0)
        pthread_cond_wait(&(pool->sync_handle->progress_condition), &(pool->sync_handle->progress_mutex));
    pthread_mutex_unlock(&(pool->sync_handle->progress_mutex));
}

问题分析与排查建议 #

1. 负载分配不稳定问题 #

当前任务队列是**后进先出（LIFO）**的栈结构（通过cursor-1取最后一个元素），这种结构下，第一个抢到锁的线程会连续取走所有任务——因为它解锁后会立刻重新尝试抢锁，而操作系统线程调度可能优先让同一个线程重复获得锁，导致其他线程抢不到任务。

解决方法：

• 将队列改为**先进先出（FIFO）**结构：取data[0]并调整后续元素位置，或用环形队列实现，避免线程连续抢占所有任务。
• 在WorkerThread_do_job返回true后，加入sched_yield()主动让渡CPU，增加其他线程抢锁的机会。

2. 偶发段错误（偏移量未完全初始化） #

核心疑点是线程同步逻辑存在漏洞，导致主线程在部分任务未完成时就进入ST阶段，或任务处理时存在数据竞争。

潜在问题与排查步骤：

1. 检查任务处理函数的内存写入：确认job_process中每个线程负责的偏移量区域完全独立，无重叠写入。若有共享写入，必须加锁或用原子操作保护。
2. 修复同步逻辑：ThreadPool_waiton_workers中先等待队列空、再等待work_in_progress为0的两步之间存在窗口——队列空后可能仍有线程在处理任务，主线程提前进入ST阶段会读取未初始化数据。可去掉等待队列空的步骤，仅等待work_in_progress为0，确保所有任务完成后再执行ST代码。
3. 用内存检测工具排查：使用valgrind的helgrind或drd工具检测数据竞争和同步错误，这类工具能定位偶发的同步漏洞。
4. 预初始化偏移量内存：MT代码执行前，将偏移量数组初始化为已知无效值（如0或-1），ST代码先检查所有值是否已正确初始化，避免越界访问。
5. 优化work_in_progress更新：当前用锁保护work_in_progress的增减，可改用原子操作（如__sync_add_and_fetch）替代，减少锁竞争的同时避免潜在锁问题。

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