如何优化基于无锁队列的多线程素数计数性能（内存受限）

问题背景 #

需实现多线程结合无锁队列的高性能素数计数，要求不得修改朴素素数检查函数，内存占用不超过1.8MB。首次尝试多线程时遇内存问题，改用链表后尝试无锁队列，但目前性能陷入瓶颈，需进一步优化从标准输入读取大量数字并统计素数的耗时。

当前实现代码 #

#include <stdio.h>
#include <stdbool.h>
#include <pthread.h>
#include <stdatomic.h>
#include <unistd.h>

#define MAX_ARRAY_SIZE 450000 // 4*450000=1.8MB，严格符合内存限制

int array[MAX_ARRAY_SIZE];
atomic_int shared_counter = ATOMIC_VAR_INIT(0);
int total_primes = 0;

// 禁止修改的朴素素数检查函数
bool isPrime(int n)
{
    if (n <= 1)
    {
        return false;
    }
    for (int i = 2; i * i <= n; i++)
    {
        if (n % i == 0)
        {
            return false;
        }
    }
    return true;
}

// 工作线程函数
void *worker(void *arg)
{
    int local_counter = 0;
    int index;

    while ((index = atomic_fetch_sub(&shared_counter, 1)) > 0)
    {
        if (isPrime(array[index]))
        {
            local_counter++;
        }
    }

    return (void *)(long long)local_counter; // 避免64位系统指针截断问题
}

int main()
{
    int num;
    int cpu_count = sysconf(_SC_NPROCESSORS_ONLN); // 动态获取CPU核心数
    pthread_t threads[cpu_count];

    // 从标准输入读取数字到数组
    int array_size = 0;
    while (scanf("%d", &num) != EOF && array_size < MAX_ARRAY_SIZE)
    {
        array[array_size++] = num;
    }

    atomic_store(&shared_counter, array_size);

    // 创建与核心数匹配的线程
    for (int i = 0; i < cpu_count; i++)
    {
        pthread_create(&threads[i], NULL, worker, NULL);
    }

    // 等待线程完成并累加结果
    for (int i = 0; i < cpu_count; i++)
    {
        void *result;
        pthread_join(threads[i], &result);
        total_primes += (int)(long long)result;
    }

    printf("%d total primes.\n", total_primes);

    return 0;
}

关键优化点 #

• 消除原子操作竞争：原代码通过atomic_fetch_sub让线程抢索引，会引发严重的缓存行竞争。建议改为静态任务划分：按CPU核心数将数组分割成固定区间，每个线程处理专属区间，完全避免同步开销，这是性能提升最显著的优化。
• 动态匹配线程数：不再硬编码线程数量，通过sysconf(_SC_NPROCESSORS_ONLN)获取当前系统CPU核心数，让线程数与核心数一致，避免过度调度导致的性能损耗。
• 严格控制内存：将MAX_ARRAY_SIZE调整为450000，单int占4字节，总内存刚好1.8MB，满足内存限制要求。
• 修复返回值问题：原代码直接返回(void *)local_counter，在64位系统会出现整数转指针的截断错误，改为(void *)(long long)local_counter确保数据完整性。

进阶优化（结合无锁队列） #

如果要实现流式处理（无需预存全部数据，进一步降低内存占用），可采用轻量无锁队列方案：

1. 实现基于链表的无锁队列，每个节点存储一个输入整数，动态分配节点但控制总内存不超限制。
2. 主线程负责从标准输入读取数字，放入无锁队列。
3. 工作线程从队列取数进行素数检查，统计本地计数。
4. 主线程读取完所有数据后，向队列放入结束标记，线程收到标记后退出，最后汇总所有线程的计数。

该方案内存占用更低，任务分发更平滑，适合处理超大输入数据。

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