并行归并排序 C++ 中的性能问题

Question

我尝试使用线程和模板编写合并排序的并行实现。 相关代码如下。

我已经将性能与 C++ STL 中的排序进行了比较。当没有产生线程时，我的代码比 std::sort 慢 6 倍。使用变量 maxthreads（和/或 FACTOR）我只能将性能提高一倍，因此在最好的情况下，我的速度比 std::sort 慢 3 倍。我已经在 16 核多处理器机器上尝试过代码。

htop 显示内核按预期使用，但为什么性能不足，我感觉不到整体运行时的并行性？

有错误吗？

感谢您的回复。

#define FACTOR 1
static unsigned int maxthreads = FACTOR * std::thread::hardware_concurrency();

unsigned int workers=0;
std::mutex g_mutex;

template <typename T>
std::vector<T>* mergesort_inplace_multithreading(
    typename std::vector<T>::iterator* listbegin,
    typename std::vector<T>::iterator *listend,
    std::vector<T>* listarg)
{
    if (*listbegin == *listend)
    {
        return listarg;
    }
    else if (*listend == *listbegin + 1)
    {
        return listarg;
    }
    else
    {
        size_t offset = std::distance(*listbegin, *listend)/2;
        typename std::vector<T>::iterator listhalf = *listbegin + offset;
        g_mutex.lock();
        if (::workers <= maxthreads-2 and maxthreads >=2)
        {
            workers += 2;

            g_mutex.unlock();

            std::thread first_thread(mergesort_inplace_multithreading<T>, listbegin, &listhalf, listarg);
            std::thread second_thread(mergesort_inplace_multithreading<T>, &listhalf, listend, listarg);
            first_thread.join();
            second_thread.join();
            g_mutex.lock();
            workers -= 2;
            g_mutex.unlock();
        }
        else
        {
            g_mutex.unlock();
            mergesort_inplace_multithreading<T>(listbegin, &listhalf, listarg);
            mergesort_inplace_multithreading<T>(&listhalf, listend, listarg);
        }

        typename std::vector<T> result;
        typename std::vector<T>::iterator lo_sorted_it = *listbegin;
        typename std::vector<T>::iterator hi_sorted_it = listhalf;
        typename std::vector<T>::iterator lo_sortedend = listhalf;
        typename std::vector<T>::iterator hi_sortedend = *listend;
        while (lo_sorted_it != lo_sortedend and hi_sorted_it != hi_sortedend)
        {
            if (*lo_sorted_it <= *hi_sorted_it)
            {
                result.push_back(*lo_sorted_it);
                ++lo_sorted_it;
            }
            else
            {
                result.push_back(*hi_sorted_it);
                ++hi_sorted_it;
            }

        }//end while

        if (lo_sorted_it != lo_sortedend)
        {
            //assert(hi_sorted_it == hi_sortedend);
            result.insert(result.end(), lo_sorted_it, lo_sortedend);
        }
        else
        {
            //assert(lo_sorted_it == lo_sortedend);
            result.insert(result.end(), hi_sorted_it, hi_sortedend);
        }
        std::copy(result.begin(), result.end(), *listbegin);
        return listarg;
    }
}

int main()
{
    //some tests
}

Answer 1

并行归并排序不需要互斥锁。而且您当然不需要为每个分区拆分启动两个线程。你启动一个线程；第二个分区在 current 线程上处理；与一个只等待另外两个线程完成的线程相比，线程资源的使用要好得多。

首先，简单的测试程序，对 2000 万个无符号整数进行排序。注意：所有使用 Apple LLVM 版本 5.1 (clang-503.0.40)（基于 LLVM 3.4svn）、64 位、posix 线程和设置为 O2 的优化编译的程序

测试程序

int main()
{
    using namespace std::chrono;

    std::random_device rd;
    std::mt19937 rng(rd());
    std::uniform_int_distribution<unsigned int> dist(0, std::numeric_limits<unsigned int>::max());

    std::vector<unsigned int> v, back(20*1000000);

    for (int i=0; i<5; ++i)
    {
        std::cout << "Generating...\n";
        std::generate_n(back.begin(), back.size(), [&](){return dist(rng);});

        time_point<system_clock> t0, t1;

        v = back;
        std::cout << "std::sort: ";
        t0 = system_clock::now();
        std::sort(v.begin(), v.end());
        t1 = system_clock::now();
        std::cout << duration_cast<milliseconds>(t1-t0).count() << "ms\n";

        v = back;
        std::cout << "mergesort_mt1: ";
        t0 = system_clock::now();
        mergesort_mt1(v.begin(), v.end());
        t1 = system_clock::now();
        std::cout << duration_cast<milliseconds>(t1-t0).count() << "ms\n";
    }

    return 0;
}

---

并行合并排序

我们从超基本的东西开始。我们将并发线程的数量限制为标准库中报告的硬件并发。一旦达到这个限制，我们就会停止发布新线程并简单地递归我们现有的线程。这种微不足道的算法一旦分布在硬件支持的线程中，就会表现出令人惊讶的良好行为。

template<typename Iter>
void mergesort_mt1(Iter begin, Iter end,
                  unsigned int N = std::thread::hardware_concurrency()/2)
{
    auto len = std::distance(begin, end);
    if (len < 2)
        return;

    Iter mid = std::next(begin, len/2);
    if (N > 1)
    {
        auto fn = std::async(mergesort_mt1<Iter>, begin, mid, N-2);
        mergesort_mt1(mid, end, N-2);
        fn.wait();
    }
    else
    {
        mergesort_mt1(begin, mid, 0);
        mergesort_mt1(mid, end, 0);
    }

    std::inplace_merge(begin, mid, end);
}

输出

Generating...
std::sort: 1902ms
mergesort_mt1: 1609ms
Generating...
std::sort: 1894ms
mergesort_mt1: 1584ms
Generating...
std::sort: 1881ms
mergesort_mt1: 1589ms
Generating...
std::sort: 1840ms
mergesort_mt1: 1580ms
Generating...
std::sort: 1841ms
mergesort_mt1: 1631ms

这看起来很有希望，但肯定可以改进。

---

并行合并+标准库排序

std::sort 算法的实现因供应商而异。该标准的主要限制是它必须具有平均复杂度 O(NlogN)。为了在性能方面实现这一点，许多std::sort 算法是标准库中最复杂、最优化的代码。我仔细阅读了一些具有多个内部排序特征的实现。我见过的一个这样的实现使用introsort（quicksort 直到递归深度受到限制，然后heapsort）用于更大的分区，一旦达到小分区，就会屈服于巨大的手动展开的 16 槽 @987654324 @。

关键是，标准库的作者明白，一种通用的排序算法根本不适合所有人。几个人经常被雇用来完成这项任务，经常一起和谐地工作。不要天真地认为你可以打败他们；而是加入他们，利用他们的辛勤工作。

修改我们的代码很简单。对于小于 1025 的所有分区，我们使用std::sort。其余部分相同：

template<typename Iter>
void mergesort_mt2(Iter begin, Iter end,
                   unsigned int N = std::thread::hardware_concurrency())
{
    auto len = std::distance(begin, end);
    if (len <= 1024)
    {
        std::sort(begin,end);
        return;
    }

    Iter mid = std::next(begin, len/2);
    if (N > 1)
    {
        auto fn = std::async(mergesort_mt2<Iter>, begin, mid, N-2);
        mergesort_mt2(mid, end, N-2);
        fn.wait();
    }
    else
    {
        mergesort_mt2(begin, mid, 0);
        mergesort_mt2(mid, end, 0);
    }

    std::inplace_merge(begin, mid, end);
}

将我们的新测试用例添加到测试程序后，我们得到：

输出

Generating...
std::sort: 1930ms
mergesort_mt1: 1695ms
mergesort_mt2: 998ms
Generating...
std::sort: 1854ms
mergesort_mt1: 1573ms
mergesort_mt2: 1030ms
Generating...
std::sort: 1867ms
mergesort_mt1: 1584ms
mergesort_mt2: 1005ms
Generating...
std::sort: 1862ms
mergesort_mt1: 1589ms
mergesort_mt2: 1001ms
Generating...
std::sort: 1847ms
mergesort_mt1: 1578ms
mergesort_mt2: 1009ms

好的。 现在我们看到了一些令人印象深刻的东西。但是我们可以挤出更多吗？

---

并行合并+带有限递归的标准排序

如果你想一想，为了充分利用std::sort 给出的所有辛勤工作，一旦我们达到满线程人口，我们就可以简单地停止递归。如果发生这种情况，只需使用std::sort 对我们拥有的任何内容进行排序，并在完成后将它们合并在一起。很难相信，这实际上会降低代码的复杂性。我们的算法变成了一种简单的跨核心分布分区，每个核心都由std::sort 处理：

template<typename Iter>
void mergesort_mt3(Iter begin, Iter end,
                   unsigned int N = std::thread::hardware_concurrency()/2)
{
    auto len = std::distance(begin, end);
    if (len <= 1024 || N < 2)
    {
        std::sort(begin,end);
        return;
    }

    Iter mid = std::next(begin, len/2);
    auto fn = std::async(mergesort_mt3<Iter>, begin, mid, N-2);
    mergesort_mt3(mid, end, N-2);
    fn.wait();
    std::inplace_merge(begin, mid, end);
}

再一次，在将它添加到我们的测试循环之后......

输出

Generating...
std::sort: 1911ms
mergesort_mt1: 1656ms
mergesort_mt2: 1006ms
mergesort_mt3: 802ms
Generating...
std::sort: 1854ms
mergesort_mt1: 1588ms
mergesort_mt2: 1008ms
mergesort_mt3: 806ms
Generating...
std::sort: 1836ms
mergesort_mt1: 1580ms
mergesort_mt2: 1017ms
mergesort_mt3: 806ms
Generating...
std::sort: 1843ms
mergesort_mt1: 1583ms
mergesort_mt2: 1006ms
mergesort_mt3: 853ms
Generating...
std::sort: 1855ms
mergesort_mt1: 1589ms
mergesort_mt2: 1012ms
mergesort_mt3: 798ms

如前所述，对于任何 1024 个或更小的分区，我们只需委托给 std::sort。如果分区更大，我们引入一个新线程来处理分割分区的一侧，使用当前线程来处理另一侧。一旦我们使线程限制 N 饱和，我们就会停止拆分，并且无论如何都简单地将所有内容委托给std::sort。简而言之，我们是std::sort 的多线程分发前端。

---

总结

我们可以在房间里发射更多子弹（使用一些元编程并假设一个固定的并发池编号），但我留给你。

你可以显着提高你的排序性能，如果你只专注于分区，分配到线程直到被点击，对地板分区使用高度优化的排序算法，然后把东西拼接在一起完成工作.是否还有改进的余地？当然。但在上面介绍的最简单的形式中，没有锁定、没有互斥体等。最终样本和裸 std::sort 之间的差异是，在 2011 年中期具有 4GB RAM 的微不足道的小型 MacBook Air 上，相同数据集的改进高达 58%和双核 i7 处理器。这令人印象深刻，考虑到它只需要很少的代码，简直太棒了真棒。

Answer 2

感谢您的回复。

互斥锁只保护无符号整数工作者（一个全局变量），它跟踪产生了多少线程。如果达到最大值（由 maxthreads 给出），则不再生成线程。 您可以使用 mergesort_mt2 中的参数 N 来实现。

你的机器有多少个内核？

仍然只有的性能似乎翻了一番......