C OpenMP 并行快速排序

在 C++ 中使用 openMP 时，我再次陷入困境。这次我尝试实现并行快速排序。

Code:

#include <iostream>
#include <vector>
#include <stack>
#include <utility>
#include <omp.h>
#include <stdio.h>

#define SWITCH_LIMIT 1000

using namespace std;

template <typename T>
void insertionSort(std::vector<T> &v, int q, int r)
{
    int key, i;
    for(int j = q + 1; j <= r; ++j)
    {
        key = v[j];
        i = j - 1;
        while( i >= q && v[i] > key )
        {
            v[i+1] = v[i];
            --i;
        }
        v[i+1] = key;
    }
}

stack<pair<int,int> > s;

template <typename T>
void qs(vector<T> &v, int q, int r)
{
    T pivot;
    int i = q - 1, j = r;
    //switch to insertion sort for small data
    if(r - q < SWITCH_LIMIT) 
    {
        insertionSort(v, q, r);
        return;
    }

    pivot = v[r];
    while(true)
    {
        while(v[++i] < pivot);
        while(v[--j] > pivot);
        if(i >= j) break;
        std::swap(v[i], v[j]); 
    }
    std::swap(v[i], v[r]);

    #pragma omp critical
    {
        s.push(make_pair(q, i - 1));
        s.push(make_pair(i + 1, r));        
    }
}

int main()
{
    int n, x;
    int numThreads = 4, numBusyThreads = 0;
    bool *idle = new bool[numThreads];
    for(int i = 0; i < numThreads; ++i)
        idle[i] = true;
    pair<int, int> p;
    vector<int> v;
    cin >> n;
    for(int i = 0; i < n; ++i)
    {
        cin >> x;
        v.push_back(x);
    }
    cout << v.size() << endl;
    s.push(make_pair(0, v.size()));

    #pragma omp parallel shared(s, v, idle, numThreads, numBusyThreads, p) 
    {
        bool done = false;
        while(!done) 
        {
            int id = omp_get_thread_num();
            #pragma omp critical
            {
                if(s.empty() == false && numBusyThreads < numThreads) 
                {
                    ++numBusyThreads;
                    //the current thread is not idle anymore
                    //it will get the interval [q, r] from stack
                    //and run qs on it
                    idle[id] = false;
                    p = s.top();                    
                    s.pop();
                }
                if(numBusyThreads == 0)
                {
                    done = true;
                }
            }
            if(idle[id] == false)
            {

                qs(v, p.first, p.second);
                idle[id] = true;
                #pragma omp critical 
                --numBusyThreads;
            }

        }
    }
    return 0;
}

算法：

为了将 openMP 用于递归函数，我使用堆栈来跟踪 qs 函数应运行的下一个间隔。我手动添加第一个间隔 [0, size]，然后在堆栈中添加新间隔时让线程开始工作。

问题：

程序结束得太早，如果您查看代码，则在创建第一组间隔（[q，i - 1]，[i + 1，r]）后不对数组进行排序。我的猜测是获得工作的线程，考虑默认情况下共享的快速排序函数的局部变量（代码中的 qs ），因此它们将它们搞乱并且在堆栈中不添加间隔。

我如何编译：

g++ -o qs qs.cc -Wall -fopenmp

我如何跑步：

./qs < in_100000 > out_100000

其中 in_100000 是一个文件，第一行包含 100000，下一行包含 100k 个整数，并以空格分隔。

我在 Linux 上使用 gcc 4.5.2

感谢您的帮助，

Dan

我实际上并没有运行你的代码，但我立即看到一个错误p，这应该是private not shared。并行调用qs: qs(v, p.first, p.second);将举行比赛p，导致不可预测的行为。局部变量位于qs应该没问题，因为所有线程都有自己的堆栈。然而，总体方法是好的。你走在正确的轨道上。

以下是我对并行快速排序的实现的一般评论。快速排序本身就是尴尬地平行，这意味着不需要同步。的递归调用qs在分区数组上是令人尴尬的并行。

然而，并行性暴露在递归的形式。如果您只是使用nestedOpenMP 中的并行性，最终会在一秒钟内拥有数千个线程。不会获得加速。因此，大多数情况下您需要将递归算法转变为交互式算法。然后，您需要实现一种工作队列。这就是你的方法。而且，这并不容易。

对于您的方法，有一个很好的基准：OmpSCR。您可以在以下位置下载http://sourceforge.net/projects/ompscr/

在基准测试中，有多个版本的基于 OpenMP 的快速排序。其中大多数与您的类似。然而，为了增加并行性，必须最小化全局队列上的争用（在您的代码中，它是s）。因此，可能有一些优化，例如拥有本地队列。尽管算法本身是纯粹并行的，但实现可能需要同步工件。而且，最重要的是，很难获得加速。

但是，您仍然可以通过两种方式直接在 OpenMP 中使用递归并行性：(1) 限制线程总数，以及 (2) 使用 OpenMP 3.0 的task.

这是第一种方法的伪代码（这仅基于 OmpSCR 的基准）：

void qsort_omp_recursive(int* begin, int* end)
{
  if (begin != end) {
    // Partition ...

    // Throttling
    if (...)  {
      qsort_omp_recursive(begin, middle);
      qsort_omp_recursive(++middle, ++end);
    } else {

#pragma omp parallel sections nowait
      {
#pragma omp section
        qsort_omp_recursive(begin, middle);
#pragma omp section
        qsort_omp_recursive(++middle, ++end);
      }
    }
  }
}

为了运行此代码，您需要调用omp_set_nested(1) and omp_set_num_threads(2)。代码非常简单。我们只是在工作划分上产生两个线程。然而，我们插入了一个简单的限制逻辑来防止过多的线程。请注意，我的实验显示这种方法的速度相当不错。

最后，您可以使用 OpenMP 3.0task，其中任务是逻辑上并发的工作。在上述所有 OpenMP 方法中，每个并行构造都会产生两个physical线程。您可能会说任务与工作线程之间存在严格的一对一映射。然而，task将逻辑任务和工作人员分开。

因为OpenMP 3.0还没有流行，所以我会使用西尔克加，这非常适合表达这种嵌套和递归并行性。在 Cilk Plus 中，并行化非常简单：

void qsort(int* begin, int* end)
{
  if (begin != end) {
    --end;
    int* middle = std::partition(begin, end,
      std::bind2nd(std::less<int>(), *end));
    std::swap(*end, *middle);

    cilk_spawn qsort(begin, middle);
    qsort(++middle, ++end);
    // cilk_sync; Only necessay at the final stage.
  }
}

我从 Cilk Plus 的示例代码中复制了此代码。您将看到一个关键字cilk_spawn是并行快速排序的一切。我将跳过 Cilk Plus 和 spawn 关键字的解释。然而，它很容易理解：两个递归调用被声明为逻辑上并发的任务。每当递归发生时，就会创建逻辑任务。但是，Cilk Plus 运行时（它实现了高效的工作窃取调度程序）将处理各种脏作业。它以最佳方式对并行任务进行排队并映射到工作线程。

请注意，OpenMP 3.0 的task本质上与 Cilk Plus 的方法类似。我的实验表明，相当不错的加速是可行的。我在 8 核机器上获得了 3~4 倍的加速。而且，加速是规模化的。 Cilk Plus 的绝对加速比 OpenMP 3.0 更高。

Cilk Plus（和 OpenMP 3.0）的方法和您的方法本质上是相同的：并行任务和工作负载分配的分离。然而，有效实施却非常困难。例如，您必须减少争用并使用无锁数据结构。