最大单卖利润-并行化版本

Question

我正在尝试使用OpenMP API（或pthread）并行化以下代码，其时间复杂度为O(n)。我想知道是否可以将输入数组分成X块（X =线程数），并为每个块并行处理。

这是一个非常经典的算法问题，到目前为止我没有看到任何人尝试实现并行化版本。

重要提示：简单的归约不能解决此问题，因为我仅从左到右读取数组。因此，并行化不是那么容易...

 #include<stdio.h>

/* The function assumes that there are at least two
   elements in array.
   The function returns a negative value if the array is
   sorted in decreasing order.
   Returns 0 if elements are equal  */
int maxDiff(int arr[], int arr_size)
{
  int max_diff = arr[1] - arr[0];
  int min_element = arr[0];
  int i;
  for(i = 1; i < arr_size; i++)
  {       
    if(arr[i] - min_element > max_diff)                               
      max_diff = arr[i] - min_element;
    if(arr[i] < min_element)
         min_element = arr[i];                     
  }
  return max_diff;
}

- user1197918

---

当我们读取数组时，你必须关注方向。我是从左到右读取的。 - user1197918

好的 - 所以这比最大差异要微妙一些，你是在寻找一个项目和随后的项目之间的最大差异，对吗？ - Jonathan Dursi

@jonathan 是的，完全正确。 - user1197918

更新的版本似乎做了正确的事情，但问题的额外限制将进一步限制扩展能力。 - Jonathan Dursi

Answer 1

由于数据依赖和低的计算需求，这很难在多核处理器中提高速度。但是，您可以通过在数组的每个块中计算本地最小值、最大值和最好的本地区域来实现一些效果，然后在块之间进行比较。由于最终步骤，运行时间为O(N) + O(P²)，进一步限制了可扩展性。

#include <stdio.h>
#include <stdlib.h>
#include <time.h>
#include <sys/time.h>
#include <limits.h>
#include <omp.h>

void tick(struct timeval *t);
double tock(const struct timeval * const t);

unsigned int maxDiff(const int * const arr, const int arr_size)
{
  int max_diff = arr[1] - arr[0];
  int min_element = arr[0];
  int i;
  for(i = 1; i < arr_size; i++)
  {
    if(arr[i] - min_element > max_diff)
      max_diff = arr[i] - min_element;
    if(arr[i] < min_element)
         min_element = arr[i];
  }
  return max_diff;
}

unsigned int ompMaxDiff(const int * const arr, const int arr_size)
{
  int nthreads=omp_get_max_threads();
  int maxes[nthreads];
  int mins [nthreads];
  unsigned int best = 0;

  for (int i=0; i<nthreads; i++) {
    mins [i] = INT_MAX;
    maxes[i] = INT_MIN;
  }

  #pragma omp parallel num_threads(nthreads) default(none) shared(mins, maxes) reduction(max:best) 
  {
      int idx = omp_get_thread_num();
      int min = INT_MAX, max = INT_MIN;

      #pragma omp for schedule(static) 
      for(int i=0; i<arr_size; i++) {
        if (arr[i] < min) min=arr[i];
        if (arr[i] > max) max=arr[i];
        if ((arr[i] - min) > best) best = arr[i] - min;
      }

      mins [idx] = min;
      maxes[idx] = max;
  }

  for (int i=0; i<nthreads-1; i++)
    for (int j=i+1; j<nthreads; j++)
        if ((maxes[j] - mins[i]) > best) best = maxes[j]-mins[i];

  return best;
}

int main(int argc, char **argv) {
    const int nitems=1000000;
    int *data = malloc(nitems*sizeof(int));

    srand(time(NULL));
    for (int i=0; i<nitems; i++)
        data[i] = rand() % 500;    /* numbers between 0 and 500 */


    data[(nitems/2)+1] = -700;
    data[(nitems/2)]   = 700;      /* a trick! shouldn't get 1400, */
                                   /* should get <= 1200 */

    struct timeval start;
    tick(&start);
    unsigned int res = maxDiff(data, nitems);
    double restime = tock(&start);

    printf("Serial: answer = %u, time = %lf\n", res, restime);

    tick(&start);
    res = ompMaxDiff(data, nitems);
    restime = tock(&start);

    printf("OpenMP: answer = %u, time = %lf\n", res, restime);

    free(data);

    return 0;
}

void tick(struct timeval *t) {
    gettimeofday(t, NULL);
}

double tock(const struct timeval * const t) {
    struct timeval now;
    gettimeofday(&now, NULL);
    return (double)(now.tv_sec - t->tv_sec) + ((double)(now.tv_usec - t->tv_usec)/1000000.);
}

运行在 8 核心上会有以下效果：

$ gcc -fopenmp -O3 -Wall -std=c11 maxdiff.c -o maxdiff
$ ./maxdiff 
Serial: answer = 1199, time = 0.001760
OpenMP: answer = 1199, time = 0.000488

- Jonathan Dursi

---

我正要给你提供修复和正确的解决方案！此外，OpenMP中的缩减被认为是可结合和可交换的。然而，OPs函数不是可交换的，但仍然是可结合的（例如，像矩阵乘法）。在这种情况下，您必须按线程存储结果（就像您所做的那样），然后串行操作。对于不可交换的操作，无法并行化。 - Z boson

糟糕 - 抱歉，@Zboson :)。另一方面，这个版本可能可以提供显著的改进。例如，事实证明（谷歌“最大和子序列并行”，这是一个可以转化为该问题的问题），通过在每个并行区域中进行更多计算，您可以将组合时间从O（P ²）提高到O（P）。但对于可能有帮助的P的小范围而言，我强烈怀疑额外的工作是否值得。 - Jonathan Dursi

一般来说，我会假设 n >> p^2，因此串行部分应该可以忽略不计。我的意思是，在 NUMA 系统上最大的 p 是多少？ - Z boson

我不确定我理解你关于静态调度的论点。你担心它可能不会按照线程号递增分配块吗？我在这方面提出了一个问题https://dev59.com/PnbZa4cB1Zd3GeqPJLoZ - Z boson

@Zboson - 哦，我没有意识到在未指定块大小的情况下保证如此强大；感谢你指出这一点，今天我学到了有用的东西。至于P，现在购买30-40个核心节点已经相当容易了，而40^2是相当可观的（但可能仍然是次要的，正如你所指出的那样）。 - Jonathan Dursi

Answer 2

我不确定具体的OpenMP，但这是一个适合并行处理的问题的关联运算符。

struct intermediate {
    int min_elem;
    int max_elem;
    int max_diff;
};

使用此函数准备单例列表。

struct intermediate singleton(int x) {
    return (struct intermediate){x, x, INT_MIN};
}

使用此函数将两个相邻的中间结果组合在一起。

struct intermediate combine(struct intermediate a, struct intermediate b) {
    return (struct intermediate){min(a.min_elem, b.min_elem),
                                 max(a.max_elem, b.max_elem),
                                 max(max(a.max_diff, b.max_diff),
                                     b.max_elem - a.min_elem)};
}

可以采用一种可能的评估策略，如下所示。

        C
       / \
      C   \
     / \   \
    /   \   \
   /     \   \
  C       C   \
 / \     / \   \
S   S   S   S   S
|   |   |   |   |
0   1   2   3   4

这里C表示组合，S表示单例。由于组合是可结合的，任何二叉树都可以。这里有另一种策略。

        C
       / \
      /   \
     /     \
    /       C
   /       / \
  C       /   C
 / \     /   / \
S   S   S   S   S
|   |   |   |   |
0   1   2   3   4

- David Eisenstat

---

@Jeb11 你可以对一个四元素数组进行类似 combine(combine(singleton(a[0]), singleton(a[1])), combine(singleton(a[2]), singleton(a[3]))) 的评估。由于 combine 是可结合的，因此在组合顺序方面有很大的灵活性。 - David Eisenstat

我只从左到右读取数组，所以你的解决方案不起作用。 - user1197918

@Jeb11 这不是Jonathan正在使用的减少方法。你试过了吗？ - David Eisenstat