使用 OpenMP 并发更新 C 中的动态数组

Question

假设许多线程尝试在动态分配数组的末尾追加元素。如果没有足够的空间，数组必须重新分配，但是它在内存中的地址可能会改变，其他线程必须知道这个改变。假设我们有以下顺序代码：

int capacity = 0;
int len = 0;
double *A = NULL;

for (int i = 0; i < n; i++) {
    /* get a stuff array with impredictably many elements */
    int k = foo(i);     // determine number of elements
    double stuff[k];
    bar(i, k, stuff);   // fill the stuff array
    
    /* Append stuff at the end of A */

    if (len + k > capacity) {  // not enough space in A? Realloc twice the size
        capacity = 2*capacity + k;
        A = realloc(A, capacity * sizeof(*A));  // pointer may change
    }
    for (int j = 0; j < k; k++)  // copy
        A[len++] = stuff[j];
}

我想并行化循环的迭代。 foo 和 bar 是线程安全的。我想尽量减少临界区的使用。特别是，我想把复制循环排除在外。

一个可能的解决方案是使用一个公平读写器锁。线程将写入-获取锁以重新分配数组，并在复制循环期间持有读取锁。如果一个线程等待写锁，它应该阻止其他线程获取读锁。

在纯 OpenMP 中执行此操作的最佳方法是什么？

Answer 1

没有完整的代码很难做出精确的评估。尽管如此，您可以通过将这些操作封装在 mutual exclusion 区域中来确保在评估和重新分配数组期间不会出现竞争条件。

这可以通过在数组重新分配之前添加barrier 来实现，然后是single 区域（这样只有一个线程在需要时重新分配数组A）。 single 子句的末尾有一个隐式屏障，因此所有线程都会等待数组重新分配。

#pragma omp parallel
{
    #pragma omp for 
    for (int i = 0; i < n; i++) {
        /* get a stuff array with impredictably many elements */
        int k = foo(i);     // determine number of elements
        double stuff[k];
        bar(i, k, stuff);   // fill the stuff array
    
        #pragma omp barrier
        #pragma omp single
        {
           if (len + k > capacity) 
           {
               capacity = 2*capacity + k;
               A = realloc(A, capacity * sizeof(*A)); 
           }
        }
        for (int j = 0; j < k; k++)  // copy
           A[len++] = stuff[j];
    } 
    // If n does not evenly divided among threads
    int rest = n % total_threads;
    if(rest != 0){
       int thread_id = omp_get_thread_num();
       if(total_threads - rest >= thread_id)
       {
          #pragma omp barrier
          #pragma omp barrier
       }
    }
}

这种方法的缺点是您只能在#pragma omp for 上使用static 调度程序；因为我们在循环中调用#pragma omp barrier，所以我们需要确保所有线程调用该屏障的次数相同。因此，为什么在循环结束后我们还需要弄清楚 for 循环的迭代次数是否在线程之间平均分配（即如果所有线程都有barrier) 调用的次数相同。

int rest = n % total_threads;
if(rest != 0){
   int thread_id = omp_get_thread_num();
   if(total_threads - rest >= thread_id)
   {
      #pragma omp barrier
      #pragma omp barrier
   }
}

如果没有，我们需要确保 missing 线程相应地调用barrier。我们调用了两次屏障，因为这是每次循环迭代调用 per 的屏障数。

在上面的代码中，我假设静态分发的默认块。对于 n = 10 和 4 个线程，这意味着：

thread 0 : works with iterations 0, 1, and 2
thread 1 : works with iterations 3, 4, and 5
thread 2 : works with iterations 6, 7
thread 3 : works with iterations 8, 9

所以 ID 为 2 和 3 的线程都需要再次调用屏障。所以其余的=2，这意味着if(2 >= thread_id)。

这种方法容易出错，而且有人可能会说它是骇人听闻的。您可以尝试的替代方法（如果它们符合您的要求）

1. 是有一个数组A每个线程，每个线程可以安全地调整其数组的大小，而不必相互同步。在并行区域之后，您确保 initial 线程将所有这些数组的值相应地合并到单个数组A；

2. 尝试找出A 大小的安全上限，并相应地预先分配该数组。

IMO，最后一个选项是迄今为止最好的选项：

1. 性能更高，因为您避免了数组大小的调整和线程之间的同步；

2. 代码更具可读性且不易出错。