线程技术：小数组多次重复并行扫描的最高效实现方法

核心问题拆解 #

你的场景中，全局barrier是性能瓶颈的核心：每次10-1000微秒的短迭代里，全局barrier的线程等待、唤醒开销占比极高，甚至超过计算本身。而stencil计算的局部依赖特性（仅依赖相邻元素），正好可以用点对点局部同步替代全局barrier，且完全不需要原子操作。

具体实现方案 #

一、静态分区+线程间局部等待（无额外调度开销，推荐） #

利用数组适配缓存的特性，将数组按线程数静态划分为连续块，每个线程固定负责一块。基于stencil的邻域依赖，线程只需等待前一个线程完成其块的最后一个元素（当前块第一个元素的左邻居），即可启动计算，无需等待全局所有线程。

代码实现

#include <atomic>
#include <thread>
#include <vector>

const int num_iter = 1000;
const int array_size = 1024 * 64; // 适配缓存的尺寸
const int num_threads = 128;

struct ThreadCtx {
    int start;
    int end;
    double* curr_buf;
    double* next_buf;
    std::atomic<bool>* prev_done;
    std::atomic<bool>* self_done;
};

void worker(ThreadCtx* ctx) {
    for (int iter = 0; iter < num_iter; ++iter) {
        // 等待前序线程完成当前迭代的左邻域依赖
        if (ctx->start > 0) {
            while (!ctx->prev_done->load(std::memory_order_acquire));
        }

        // 执行3点stencil求和计算
        for (int j = ctx->start; j < ctx->end; ++j) {
            double left = (j > 0) ? ctx->curr_buf[j-1] : 0.0;
            double curr = ctx->curr_buf[j];
            double right = (j < array_size-1) ? ctx->curr_buf[j+1] : 0.0;
            ctx->next_buf[j] = left + curr + right;
        }

        // 标记当前线程完成本轮计算，通知后续线程
        ctx->self_done->store(true, std::memory_order_release);

        // 切换缓冲（无需全局swap，线程内部直接切换指针）
        if (iter < num_iter - 1) {
            std::swap(ctx->curr_buf, ctx->next_buf);
            // 重置标志位，为下一轮做准备
            if (ctx->start > 0) {
                ctx->prev_done->store(false, std::memory_order_release);
            }
            ctx->self_done->store(false, std::memory_order_release);
        }
    }
}

int main() {
    // 初始化双缓冲数组
    std::vector<double> buf1(array_size, 0.0);
    std::vector<double> buf2(array_size, 0.0);
    // 初始化线程完成标志
    std::vector<std::atomic<bool>> done_flags(num_threads, false);
    std::vector<ThreadCtx> thread_ctxs(num_threads);
    std::vector<std::thread> threads;

    // 划分数组块
    int chunk_size = array_size / num_threads;
    for (int t = 0; t < num_threads; ++t) {
        thread_ctxs[t].start = t * chunk_size;
        thread_ctxs[t].end = (t == num_threads-1) ? array_size : (t+1)*chunk_size;
        thread_ctxs[t].curr_buf = buf1.data();
        thread_ctxs[t].next_buf = buf2.data();
        thread_ctxs[t].prev_done = (t > 0) ? &done_flags[t-1] : nullptr;
        thread_ctxs[t].self_done = &done_flags[t];
        threads.emplace_back(worker, &thread_ctxs[t]);
    }

    // 等待所有线程完成
    for (auto& th : threads) {
        th.join();
    }

    return 0;
}

二、基于任务依赖的调度（适合动态负载场景） #

如果需要动态负载均衡，可以用TBB等任务库的依赖调度机制，将每个数组块的迭代任务设置为依赖前一块的上一轮任务。这样任务会在依赖完成后自动启动，无需全局barrier。

简化代码实现（TBB）

#include <tbb/task.h>

class StencilChunkTask : public tbb::task {
public:
    int start, end;
    double* in_buf;
    double* out_buf;
    int iter;
    tbb::task* dep_task;

    StencilChunkTask(int s, int e, double* in, double* out, int it, tbb::task* dep)
        : start(s), end(e), in_buf(in), out_buf(out), iter(it), dep_task(dep) {}

    tbb::task* execute() override {
        // 等待依赖任务完成
        if (dep_task) dep_task->wait_for_all();

        // 执行stencil计算
        for (int j = start; j < end; ++j) {
            double left = (j > 0) ? in_buf[j-1] : 0.0;
            double curr = in_buf[j];
            double right = (j < array_size-1) ? in_buf[j+1] : 0.0;
            out_buf[j] = left + curr + right;
        }

        // 生成下一轮任务（如果不是最后一轮）
        if (iter < num_iter - 1) {
            std::swap(in_buf, out_buf);
            auto next_task = new(allocate_child()) StencilChunkTask(start, end, in_buf, out_buf, iter+1, this);
            set_ref_count(1);
            spawn(*next_task);
            return nullptr;
        }
        return nullptr;
    }
};

// 主线程启动逻辑
int main() {
    tbb::task_group tg;
    std::vector<StencilChunkTask*> first_tasks(num_threads);
    int chunk_size = array_size / num_threads;
    std::vector<double> buf1(array_size, 0.0);
    std::vector<double> buf2(array_size, 0.0);

    for (int t = 0; t < num_threads; ++t) {
        int start = t * chunk_size;
        int end = (t == num_threads-1) ? array_size : (t+1)*chunk_size;
        tbb::task* dep = (t > 0) ? first_tasks[t-1] : nullptr;
        first_tasks[t] = new(tg.allocate_root()) StencilChunkTask(start, end, buf1.data(), buf2.data(), 0, dep);
        tg.spawn(*first_tasks[t]);
    }

    tg.wait();
    return 0;
}

关键注意点 #

1. 无需原子操作：双缓冲机制下，每轮迭代中curr_buf只读，next_buf的每个位置仅由一个线程写入，完全不存在并发写冲突，不需要原子变量或锁。
2. 缓存友好性：静态分区保证线程连续访问数组，配合你已做的缓存行填充，数据能稳定驻留L2/L3缓存。
3. 避免全局同步：无论是线程局部等待还是任务依赖，都只做必要的局部同步，彻底消除全局barrier的高额开销。

内容的提问来源于stack exchange，提问作者zx-81