TBB parallel_for复制数组非零元素结果异常，原因是什么如何修复？

• 写操作与原子计数存在时序竞争：虽然count是原子类型，但dst[count] = src[i]、twodouble[count] = 2.0 * src[i]和count++这三步不是原子操作。多个线程同时读取到相同的count值时，会往同一个下标位置写入数据，最后多次累加count，导致中间的下标位置没有被写入数据，就会出现你观察到的0值空缺。
• 代码还存在几个拼写错误会导致编译问题：tbb::paralell_for应为tbb::parallel_for，szie_t应为size_t，esp大概率是浮点数判断阈值epsilon的拼写错误。

方案1：先获取原子下标再写入（最小改动方案） #

把计数的递增操作放到写操作之前，用原子操作先拿到当前可用的下标再执行写入，即可避免写入竞争，适合非零元素数量不多的场景：

double src[NUM] = {a0, a1, a2, ..., an};
double dst[NUM] = {0}; 
double twodouble[NUM] = {0};
tbb::atomic<int> count = 0;
const double eps = 1e-9; // 定义判断非零的阈值

tbb::parallel_for(tbb::blocked_range<size_t>(0, NUM),
  [&](const tbb::blocked_range<size_t>& r){
    for (size_t i = r.begin(); i < r.end(); ++i) {
        if (std::abs(src[i]) < eps) continue;
        // 先原子递增获取当前可用下标，避免写入竞争
        int idx = count.fetch_and_add(1);
        dst[idx] = src[i];
        twodouble[idx] = 2.0 * src[i];
    }
});

方案2：分块统计+前缀和（高性能方案，适合大数据量场景） #

如果非零元素数量多，频繁的原子操作会带来较大开销，可采用分块统计+前缀和的方案，全程几乎无原子操作、无线程写冲突，性能表现更好：

double src[NUM] = {a0, a1, a2, ..., an};
double dst[NUM] = {0}; 
double twodouble[NUM] = {0};
const double eps = 1e-9;
// 块大小可按需调整，通常取1024~4096
const size_t block_size = 1024;
const size_t num_blocks = (NUM + block_size - 1) / block_size;
std::vector<int> block_counts(num_blocks, 0);

// 第一步：并行统计每个块的非零元素数量
tbb::parallel_for(size_t(0), num_blocks, [&](size_t b){
    size_t start = b * block_size;
    size_t end = std::min(start + block_size, (size_t)NUM);
    int cnt = 0;
    for(size_t i = start; i < end; i++){
        if(std::abs(src[i]) >= eps) cnt++;
    }
    block_counts[b] = cnt;
});

// 第二步：计算前缀和，得到每个块在目标数组的起始写入偏移
std::vector<int> block_offsets(num_blocks + 1, 0);
for(size_t i = 0; i < num_blocks; i++){
    block_offsets[i+1] = block_offsets[i] + block_counts[i];
}

// 第三步：并行写入目标数组
tbb::parallel_for(size_t(0), num_blocks, [&](size_t b){
    size_t start = b * block_size;
    size_t end = std::min(start + block_size, (size_t)NUM);
    int idx = block_offsets[b];
    for(size_t i = start; i < end; i++){
        if(std::abs(src[i]) >= eps){
            dst[idx] = src[i];
            twodouble[idx] = 2.0 * src[i];
            idx++;
        }
    }
});

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