multithreading - 这个 OpenMP 屏障有什么解决方法吗？

Question

我有这个用 OpenMp 编写的并行区域:

std::vector<T> sharedResult;
#pragma omp parallel
{
std::vector<T> result;
#pragma omp for nowait
for(int i=0; i<n; i++){
  //fill result
}
#pragma omp critical{
  sharedResult.insert(sharedResult.end(), result.begin(), result.end());
}
#pramga omp barrier
#pragma omp for nowait
for(size_t i=0; i<sharedResult.size(); i++){
  foo(sharedResult[i]);
}
...
}

恐怕 #pragma omp barrier是必要的。我认为的原因是，否则当一个线程到达最后一个 #pragma omp for , sharedResult.size()在那一刻仍然没有处于他的最终状态(在之前的并行完成时获得)。请注意，不幸的是 sharedResult的大小以前是未知的。

不幸的是，我注意到这个障碍会产生很大的开销，即一个特定的迭代比所有其他迭代都更昂贵，因此所有线程都必须等待执行该迭代的线程。这可以认为是负载不平衡，但我没有找到任何解决方案来解决这个问题。

所以我的问题是:有没有办法在不等待前一个完成的情况下开始最后一个并行，或者没有办法改进这个？

Answer 1

我同意屏障是必要的。我看到了几种出路，随着复杂性的增加并且可能会提高效率:

任务

为每个结果元素发布一个任务:

#pragma omp parallel
{
    std::vector<T> result;
    #pragma omp for nowait
    for(int i=0; i<n; i++){
        //fill result
    }
    // would prefer a range based loop here, but
    // there seem to be issues with passing references 
    // to tasks in certain compilers
    for(size_t i=0; i<result.size(); i++){
    {
        #pragma omp task
        foo(result[i]);
    }
}

您甚至可以在初始循环中发布任务。如果任务太多，您可能会获得大量开销。

使用已完成的线程处理结果队列

现在这个更棘手 - 特别是您需要区分结果队列为空和所有线程完成它们的第一个循环。

std::vector<T> sharedResult;
int threadsBusy;
size_t resultIndex = 0;
#pragma omp parallel
{
    #pragma omp single
    threadsBusy = omp_num_threads();

    std::vector<T> result;
    #pragma omp for nowait
    for(int i=0; i<n; i++){
        //fill result
    }

    #pragma omp critical
    {
        sharedResult.insert(sharedResult.end(), result.begin(), result.end());
        threadsBusy--;
    }

    do {
        bool hasResult, allThreadsDone;
        // We need a copy here as the vector may be resized
        // and elements may become invalid by insertion
        T myResult;
        #pragma omp critical
        {
            if (resultIndex < sharedResult.size()) {
                resultIndex++;
                hasResult = true;
                myResult = sharedResult[myResult];
            } else {
                hasResult = false;
            }
            allThreadsDone = threadsBusy == 0;
        }
        if (hasResult) {
            foo(myResult);
        } else {
            if (allThreadsDone) {
                break;
            }
            // If we just continue here, we will spin on the mutex
            // Unfortunately there are no condition variables in OpenMP
            // So instead we go for a quick nap as a compromise
            // Feel free to tune this accordingly
            std::this_thread::sleep_for(10ms);
        }
    } while (true);
}

注意:通常我会测试我在这里发布的代码，但由于缺乏完整的示例而无法测试。

通过并行循环处理结果 block

最后，对于已经完成的结果，您可以多次并行运行 for 循环。然而，这有许多问题。首先，所有线程都必须遇到每个工作共享区域，即使是较晚完成第一个工作的线程也是如此。所以你必须跟踪你运行的循环。此外，每个线程的循环绑定(bind)必须相同 - 您只能阅读 sharedResult.size()在关键部分。因此，您必须在临界区中由一个线程预先将其读取到共享变量中，但要等待所有线程，直到它被正确读取。此外，您将不得不使用动态调度，否则您可能会使用静态调度，并且您将等待最后完成的线程。您编辑的示例不执行这些操作。我不会想当然地认为 for nowait schedule(dynamic)可以在团队中的所有线程进入之前完成(但它适用于 libgomp)。综合考虑，我不会真的去那里。