C ++合并排序与线程较慢

Question

我已经用 C++ 实现了一个归并排序算法。
在算法内部，它会检查数组的大小是否大于min_size_to_thread，如果是：使用线程递归调用函数。

但是当我增加min_size_to_thread: 减少正在使用的线程数时，函数变得更快。即使从 1 个线程变为 2 个线程。

我的假设是函数速度会随着更多线程的增加而增加，然后再次开始下降。这对我来说没有任何意义，所以我开始相信我的实现在某种程度上是错误的。

template <typename T>
void merge_sort(T S[], int S_size, int min_size_to_thread)
{
    if (S_size < 2) return;

    // Left Sequence
    int L_size = S_size / 2;
    T* L = new T[L_size];
    for (int i = 0; i < L_size; i++)
    {
        L[i] = S[i];
    }

    // Right Sequence
    int R_size = (S_size + 1) / 2;
    T* R = new T[R_size];
    for (int i = 0; i < R_size; i++)
    {
        R[i] = S[i + L_size];
    }

    if (S_size > min_size_to_thread)
    {
        std::thread thread_left(merge_sort<T>, L, L_size, min_size_to_thread);
        std::thread thread_right(merge_sort<T>, R, R_size, min_size_to_thread);
        thread_right.join();
        thread_left.join();
    }
    else
    {
        merge_sort<T>(L, L_size, min_size_to_thread);
        merge_sort<T>(R, R_size, min_size_to_thread);
    }

    int S_iterator = 0;
    int L_iterator = 0;
    int R_iterator = 0;

    while ((L_iterator < L_size) && (R_iterator < R_size))
    {
        if (L[L_iterator] < R[R_iterator])
        {
            S[S_iterator] = L[L_iterator];
            ++L_iterator;
        }
        else
        {
            S[S_iterator] = R[R_iterator];
            ++R_iterator;
        }
        ++S_iterator;
    }

    while (L_iterator < L_size)
    {
        S[S_iterator] = L[L_iterator];
        ++L_iterator;
        ++S_iterator;
    }

    while (R_iterator < R_size)
    {
        S[S_iterator] = R[R_iterator];
        ++R_iterator;
        ++S_iterator;
    }

    delete[] L;
    delete[] R;
}

int main()
{
    const int S_size = 500000;
    unsigned char S[S_size];
    for (int i = 0; i < S_size; ++i)
    {
        S[i] = i % 255;
    }

    int min_size_to_thread;

    min_size_to_thread = 250;
    auto t1 = std::chrono::high_resolution_clock::now();
    merge_sort(S, S_size, min_size_to_thread);
    auto t2 = std::chrono::high_resolution_clock::now();
    std::cout << "size > " << min_size_to_thread << ": " << (t2 - t1) / std::chrono::milliseconds(1) << std::endl;

    for (int i = 0; i < S_size; ++i)
    {
        S[i] = i % 255;
    }

    min_size_to_thread = 500;
    t1 = std::chrono::high_resolution_clock::now();
    merge_sort(S, S_size, min_size_to_thread);
    t2 = std::chrono::high_resolution_clock::now();
    std::cout << "size > " << min_size_to_thread << ": " << (t2 - t1) / std::chrono::milliseconds(1) << std::endl;

    for (int i = 0; i < S_size; ++i)
    {
        S[i] = i % 255;
    }

    min_size_to_thread = 1000;
    t1 = std::chrono::high_resolution_clock::now();
    merge_sort(S, S_size, min_size_to_thread);
    t2 = std::chrono::high_resolution_clock::now();
    std::cout << "size > " << min_size_to_thread << ": " << (t2 - t1) / std::chrono::milliseconds(1) << std::endl;

    for (int i = 0; i < S_size; ++i)
    {
        S[i] = i % 255;
    }

    min_size_to_thread = 10000;
    t1 = std::chrono::high_resolution_clock::now();
    merge_sort(S, S_size, min_size_to_thread);
    t2 = std::chrono::high_resolution_clock::now();
    std::cout << "size > " << min_size_to_thread << ": " << (t2 - t1) / std::chrono::milliseconds(1) << std::endl;

    for (int i = 0; i < S_size; ++i)
    {
        S[i] = i % 255;
    }

    min_size_to_thread = 250000;
    t1 = std::chrono::high_resolution_clock::now();
    merge_sort(S, S_size, min_size_to_thread);
    t2 = std::chrono::high_resolution_clock::now();
    std::cout << "size > " << min_size_to_thread << ": " << (t2 - t1) / std::chrono::milliseconds(1) << std::endl;

    for (int i = 0; i < S_size; ++i)
    {
        S[i] = i % 255;
    }

    min_size_to_thread = 500000;
    t1 = std::chrono::high_resolution_clock::now();
    merge_sort(S, S_size, min_size_to_thread);
    t2 = std::chrono::high_resolution_clock::now();
    std::cout << "size > " << min_size_to_thread << ": " << (t2 - t1) / std::chrono::milliseconds(1) << std::endl;

    return 0;
}

Answer 1

我已经编译并运行了您的确切程序，除了添加包含之外没有任何修改，结果或多或少与您预期的一样：

size > 250: 169
size > 500: 85
size > 1000: 50
size > 10000: 29
size > 250000: 42
size > 500000: 89

根据您的屏幕截图，我推测您正在 Visual Studio 中运行您的代码。默认运行按钮会将调试器附加到您的可执行文件并降低运行时性能。相反，请按 Ctrl+F5 以在没有调试器的情况下运行，或者从菜单 Debug -> Start without Debugging 中运行。

Answer 2

我认为这是缓存的问题。具体来说，false-sharing 会减慢算法的速度，因为数据会写入多个线程之间共享的页面。 （不同的处理器内核试图跟上共享内存页面）如果 min_size_to_thread 是您的处理器的 page-size 的倍数，并且您的数组是 aligned在页面边界上，性能会提高。在这种情况下，线程之间不会共享页面。

我总是将线程的创建限制在一个恒定的数量，在四核机器上运行 100 个线程只是为了对数组进行排序是没有意义的。 由于繁重的上下文切换，在单个核心上运行多个线程会产生成本。根据我的经验，最大线程数始终是核心数乘以 2。一个核心可以处理大约 2 个线程，而无需性能成本。对于四核 CPU，程序一次最多应该运行 8 个线程。 这意味着一个算法可以创建8个子线程，父线程只是joins线程，或创建7个子线程，在父线程中运行算法的一部分，最后join另外7个线程。

总是配置文件，它可能有完全不同的原因。