OpenMP 多线程更新同一个数组

Posted

技术标签:

【中文标题】OpenMP 多线程更新同一个数组【英文标题】:OpenMP multiple threads update same array 【发布时间】:2017-08-20 19:28:00 【问题描述】:

我的程序中有以下代码,我想使用 OpenMP 加速它。

...
for(i=curr_index; i < curr_index + rx_size; i+=2) 
    int64_t tgt = rcvq[i];
    int64_t src = rcvq[i+1];
    if (!TEST(tgt)) 
        pred[tgt] = src;
        newq[newq_count++] = tgt;
    
 

目前我有一个版本如下:

...
chunk = rx_sz / omp_nthreads;

#pragma omp parallel for num_threads(omp_nthreads)
for (ii = 0; ii < omp_nthreads; ii++)  
    int start = curr_index + ii * chunk;
    for (index = start; index < start + chunk; index +=2)  
        int64_t tgt = rcvq[index];
        int64_t src = rcvq[index+1];
        if (!TEST(tgt)) 
            pred[tgt] = src;

            #pragma omp critical 
            newq[newq_count++] = tgt;
        
    

当我运行 OpenMP 版本时,我发现与原始版本相比性能大幅下降。我认为这个问题可能是因为“omp critical”阻止了并行处理。我想知道我的代码可以增强什么,这样我就可以获得比串行版本更好的性能。在代码中,rx_sz 始终是 omp_nthreads 的倍数。

【问题讨论】:

【参考方案1】:

是的,关键部分限制了您的表现。您应该在本地为每个线程收集结果,然后将它们合并。

size_t newq_offset = 0;
#pragma omp parallel

    // Figure out something clever here...
    const size_t max_newq_per_thread = max_newq / omp_get_num_threads();
    int64_t* local_newq = malloc(max_results_per_thread * sizeof(int64_t));
    size_t local_newq_count = 0;

    #pragma omp parallel for
    for (i=curr_index; i < curr_index + rx_size; i+=2)
        int64_t tgt = rcvq[2*index];
        int64_t src = rcvq[2*index+1];
        if (!TEST(tgt)) 
            pred[tgt] = src;
            local_newq_count++;
            assert(local_newq_count < max_newq_per_thread);
            local_newq[local_newq_count] = tgt;
        
    
    int local_offset;
    #pragma omp atomic capture
    
        local_offset = offset;
        offset += local_newq_count;
    
    for (size_t i = 0; i < counter; i++)
       
        res_global[i + local_offset] = res[i];
    

使用这种方法,所有线程在合并时并行工作,atomic capture 上的争用很少。注意你也可以用atomic capture做一个简单的版本,比临界区效率更高,但还是会很快成为瓶颈:

size_t newq_count_local;
#pragma omp atomic capture
newq_count_local = newq_count++;
newq[newq_count_local] = tgt;
newq 内的任何版本订单不保证 始终尽可能将变量声明为本地变量!特别是在使用 OpenMP 时。您发布的critical-版本错误,因为index(在外部范围内定义)隐式在线程之间共享。 所有这些都假定rcvq 中没有重复项。否则,您会在 pred[tgt] = src; 上获得竞争条件。 您手动分割循环的方法过于复杂。不需要做两个循环,只需使用一个pragma omp for 循环。

另一个答案是正确的。但是,它是 C++,而不是标记为 C。使用 std::vector&lt;std::vector&lt;&gt;&gt; 时还有一个微妙但显着的性能问题。通常一个向量用三个指针来实现,共24个字节。在push_back 时,指针之一递增。这意味着,a) 来自多个线程的本地向量的指针驻留在同一缓存行上,并且 b) 在每次成功的 TESTpush_back 读取和写入其他线程使用的缓存行时.这个缓存线必须一直在内核之间移动,极大地限制了这种方法的可扩展性。这称为虚假分享。

我根据另一个答案实现了一个小测试,给出了以下性能:

0.99 s - 单线程 1.58 s - 同一插槽的两个相邻内核上的两个线程 2.13 s - 不同套接字的两个内核上的两个线程 0.99 s - 两个线程共享一个内核 0.62 s - 两个套接字上有 24 个线程

而以上 C 版本的扩展性要好得多:

0.46 s - 单线程(C 与 C++ 无法比较) 0.24 s - 两个线程 0.04 s - 24 个线程

【讨论】:

【参考方案2】:

我很确定 omp 关键部分在这一点上限制了你的表现。

我建议您将结果收集到单独的缓冲区/向量中,并在并行处理完成后将它们合并(当然,如果顺序对您来说并不重要)

vector<vector<int64_t>> res;
res.resize(num_threads);
#pragma omp parallel for
for (index = 0; index < rx_sz/2; ++index)  
        int64_t tgt = rcvq[2*index];
        int64_t src = rcvq[2*index+1];
        if (!TEST(tgt)) 
            pred[tgt] = src;

            res[omp_get_thread_num()].push_back(tgt);
        
    

// Merge all res vectors if needed

【讨论】:

您的方法总体上很好,但是为此使用std::vector&lt;std::vector&lt;&gt;&gt; 存在一个微妙但重要的性能问题。我已经发布了另一个详细说明该问题的答案。 好点,但问题是如何摆脱关键部分,我已经提供了如何做到这一点的想法。由 Lemon 来做进一步的性能优化。整体性能实际上取决于TEST 的复杂性以及它发生的频率。

以上是关于OpenMP 多线程更新同一个数组的主要内容,如果未能解决你的问题,请参考以下文章

OpenMP:共享同一算法的单线程和多线程实现

openMP 的多线程崩溃

openMP多线程编程

使用 openMP 进行多核处理与多线程

NodeJS:具有多线程的本机 C++ 模块(openmp)

Openmp 多线程代码在使用多个线程时给出不同的答案