为什么多线程速度较慢？

Question

所以我正在尝试编写一个程序来查找质数。该项目的真正目的只是为了学习多线程。起初我编写了一个单线程程序，在1分钟内可以找到最高达13,633,943。我的多线程版本仅能找到10,025,627。

以下是我的单线程程序代码：

#include <iostream>

using namespace std;

bool isprime(long num)
{
    long lim = num/2;
    if(num == 1)
    {
        return 0;
    }
    for(long i = 2; i <= lim; i++)
    {
        if (num % i == 0)
        {
            return 0;
        }
        else{ lim = num/i; }
    }
    return 1;
}

int main()
{
    long lim;
    cout << "How many numbers should I test: ";
    cin >> lim;
    for(long i = 1; i <= lim || lim == 0; i++)
    {
        if(isprime(i))
        {
            cout << i << endl;
        }
    }
}

这是我的多线程程序代码。

extern"C"
{
    #include <pthread.h>
    #include <unistd.h>
}
#include <iostream>

using namespace std;

bool isprime(long num);
void * iter1(void * arg);
void * iter2(void * arg);
void * iter3(void * arg);
void * iter4(void * arg);


int main()
{
    //long lim;
    //cout << "How many numbers should I test: ";
    //cin >> lim;
    pthread_t t1;
    char mem1[4096];//To avoid false sharing. Needed anywhere else?
    pthread_t t2;
    char mem2[4096];//These helped but did not solve problem.
    pthread_t t3;
    pthread_create(&t1, NULL, iter1, NULL);
    pthread_create(&t2, NULL, iter2, NULL);
    pthread_create(&t3, NULL, iter3, NULL);
    iter4(0);
}

bool isprime(long num)
{
    long lim = num/2;
    if(num == 1)
    {
        return 0;
    }
    for(long i = 2; i <= lim; i++)
    {
        if (num % i == 0)
        {
            return 0;
        }
        else{ lim = num/i; }
    }
    return 1;
}

void * iter1(void * arg)
{
    for(long i = 1;; i = i + 4)
    {
        if(isprime(i))
        {
            cout << i << endl;
        }
    }
return 0;
}

void * iter2(void * arg)
{
    for(long i = 2;; i = i + 4)
    {
        if(isprime(i))
        {
            cout << i << endl;
        }
    }
return 0;
}

void * iter3(void * arg)
{
    for(long i = 3;; i = i + 4)
    {
        if(isprime(i))
        {
            cout << i << endl;
        }
    }
return 0;
}

void * iter4(void * arg)
{
    for(long i = 4;; i = i + 4)
    {
        if(isprime(i))
        {
            cout << i << endl;
        }
    }
return 0;
}

让我特别困惑的是，系统监视器显示单线程CPU使用率为25％，而多线程使用率为100％。这难道不应该意味着它做了四倍的计算吗？

- Dmobb Jr.

---

由于线程上下文切换的时间，有时可能会出现较慢的情况。 - Grijesh Chauhan

另外，iter2和iter4几乎没有任何关系。 - Daniel Fischer

你有多少个核心/CPU？ - Oliver Charlesworth

还有一个问题是将来自不同线程的结果输出（cout）到一个单独的线程中进行显示/控制台处理，这也需要一定的同步。 - Veger

@DmobbJr。不要为cout竞争。让每个线程将其发现的质数写入一个数组（每个线程都有自己的数组），然后在之后打印它们。这应该可以通过多线程获得加速（但仅适用于您的物理核心；如果两个线程在同一个核心上运行相似的任务，则超线程会使事情变慢）。 - Daniel Fischer

Answer 1

我相当确定cout是一个共享资源，即使它实际上能够正确地打印每个数字并按正确顺序打印，但这样做会非常拖慢速度。

我已经做过类似的事情（更加灵活，并使用原子操作“选择下一个数字”），在我的四核机器上速度几乎快了4倍。但前提是如果不打印任何内容。如果输出到控制台，则会慢得多，因为大部分时间用于移动像素而不是实际计算。

注释掉 cout << i << endl; 这一行，它将运行更快。

编辑：使用我的测试程序，带有打印：

Single thread: 15.04s. 
Four threads: 11.25s

不打印：

Single threads: 12.63s.
Four threads: 3.69s.

3.69 * 4 = 14.76秒，但我Linux机器上的time命令显示总运行时间为12.792秒，所以很明显有一些时间并不是所有线程都在运行 - 或者存在一些记账错误...

- Mats Petersson

---

注释掉 cout，编译器会优化整个函数。这样肯定会跑得更快 :-D - user405725

@VladLazarenko 我非常怀疑这个函数本身会消失，如果真的消失了，只需将您的优化退回一步即可。 - Richard J. Ross III

因此，在函数结束时将所有质数的总和计算并打印出结果。 - Mats Petersson

我非常确定每个线程只是将质数抛到标准输出，不关心顺序或任何其他事情，因此它不应该减慢速度。 - Dmobb Jr.

@DmobbJr.：我刚刚添加了一些代码计时，其中使用和不使用cout（它是相同的源代码，有一个if (verbose)来决定是否应该打印）。 - Mats Petersson

Answer 2

我认为你目前的问题很大程度上在于，你把真正能够支持多线程操作的部分（查找质数）淹没在了噪声中（将输出写入控制台所需的时间）。

为了让你对其影响有所了解，我稍微改写了你的主函数，将打印质数和查找质数分开。为了更方便地计时，我也让它从命令行接收限制条件，而不是交互式输入，代码如下：

int main(int argc, char **argv) {
    if (argc != 2) {
        std::cerr << "Usage: bad_prime <limit:long>\n";
        return 1;
    }
    std::vector<unsigned long> primes;

    unsigned long lim = atol(argv[1]);

    clock_t start = clock();

    for(unsigned long i = 1; i <= lim; i++)
        if(isprime(i))
            primes.push_back(i);
    clock_t stop = clock();

    for (auto a : primes)
        std::cout << a << "\t";

    std::err << "\nTime to find primes: " << double(stop-start)/CLOCKS_PER_SEC << "\n";
}

跳过成千上万行的质数本身，我得到了如下结果：

Time to find primes: 0.588


Real    48.206
User    1.68481
Sys     3.40082

大致需要半秒钟的时间来找到质数，超过47秒才能将其打印出来。假设真正的目的是将输出写入控制台，那么我们可能会直接停在这里。即使多线程可以完全消除查找质数的时间，我们仍然只能将最终时间从约48.2秒更改为约47.6秒--这不太值得。

因此，我暂时假设真正的目的是将输出写入类似文件之类的东西。既然费力地将代码变成多线程的形式，但每个线程都运行非常低效的代码似乎毫无意义，所以我想优化（或者至少是减少效率）单线程代码作为起点。

首先，我删除了 endl 并将其替换为 "\n"。如果将输出定向到文件，则将运行时间从0.968秒降低到0.678秒--endl 在写入换行符后还会刷新缓冲区，而该缓冲区刷新大约占程序总时间的三分之一。

在同样的基础上，我将你的isprime重写为至少稍微高效一些的东西:

bool isprime(unsigned long num) {
    if (num == 2)
        return true;

    if(num == 1 || num % 2 == 0)
        return false;

    unsigned long lim = sqrt(num);

    for(unsigned long i = 3; i <= lim; i+=2)
        if (num % i == 0)
            return false;

    return true;
}

这当然还有改进的空间（例如埃拉托色尼筛法），但它简单，直接，并且速度大约是两到三倍（上面的时间是基于使用此isprime而不是您的时间）。

目前，多线程素数查找至少有可能有意义：由于素数查找大约需要0.6秒中的0.5，即使我们只能将速度提高一倍，我们也应该看到总体时间的显着差异。

将输出与素数查找分开也为我们编写代码的多线程版本提供了更好的基础。通过使每个线程将其结果写入单独的向量，我们可以获得有意义的（非交错的）输出，而无需在cout等上进行锁定——我们分别计算每个块，然后按顺序打印出每个向量。

实现这种方式的代码可能如下所示：

#include <iostream>
#include <vector>
#include <time.h>
#include <math.h>
#include <thread>

using namespace std;

bool isprime(unsigned long num) {
    // same as above
}

typedef unsigned long UL;

struct params { 
    unsigned long lower_lim;
    unsigned long upper_lim;
    std::vector<unsigned long> results;

    params(UL l, UL u) : lower_lim(l), upper_lim(u) {}
};

long thread_func(params *p) { 
    for (unsigned long i=p->lower_lim; i<p->upper_lim; i++)
        if (isprime(i))
            p->results.push_back(i);
    return 0;
}

int main(int argc, char **argv) {
    if (argc != 2) {
        std::cerr << "Usage: bad_prime <limit:long>\n";
        return 1;
    }

    unsigned long lim = atol(argv[1]);

    params p[] = {
        params(1, lim/4),
        params(lim/4, lim/2),
        params(lim/2, 3*lim/4),
        params(3*lim/4, lim)
    };

    std::thread threads[] = {
        std::thread(thread_func, p), 
        std::thread(thread_func, p+1),
        std::thread(thread_func, p+2),
        std::thread(thread_func, p+3)
    };

    for (int i=0; i<4; i++) {
        threads[i].join();
        for (UL p : p[i].results)
            std::cout << p << "\n";
    }
}

在与之前相同的机器上运行（一台相当旧的双核处理器），我得到了以下结果：

Real    0.35
User    0.639604
Sys     0

这个算法的可扩展性非常好。如果我们只从多核计算中获益，那么我们预期找到质数所需的时间将会减少一半（我正在双核处理器上运行此代码），写入数据到磁盘的时间将保持不变（多线程并不能加快硬盘速度）。根据这些，完美的扩展将使我们的时间为0.59 / 2 + 0.1 = 0.40秒。

除此之外我们可以看到（尽管进步很小），每个线程在寻找质数的同时，线程1开始将数据写入磁盘，线程2、3和4仍在计算，以此类推，线程2开始将数据写入磁盘，线程3和4还在计算，最后线程3开始向磁盘写入数据，当线程4还在计算时。

需要补充的是，我们看到的改进可能过于微小，可能只是计时误差。我执行了单线程和多线程版本的多次测试，虽然两者都有一些变化，但多线程版本始终比计算速度提高的部分更快。

我差点忘了：为了了解这对整体速度的影响，我进行了一个测试，看看找到13,633,943以内的所有质数需要多长时间，而使用原始版本一分钟即可找到。尽管我使用的CPU几乎肯定更慢（一款约7年前的Athlon 64 X2 5200+），但此版本仅需12.7秒即可完成。

最后需要说明的是，至少目前为止，我省略了你插入的填充以防止虚假共享。根据我得到的时间，它们似乎并不需要（或有用）。

- Jerry Coffin

Answer 3

这取决于操作系统分配给您的代码运行的CPU数量。每个线程都是CPU绑定的，因此如果只有一个CPU，它将运行一个线程一段时间，分割时间片，运行下一个线程等等，这不会更快，而且可能会更慢，具体取决于线程交换的开销。至少在Solaris上，值得告诉它您希望所有线程同时运行。

我没有遇到过像其他帖子建议的那样串行化输出的实现。通常，您会获得如下输出。

235 iisi s  ppprririimmme
ee

因此，您的输出可能表明操作系统未为您分配多个线程。

您可能遇到的另一个问题是，与向文件输出相比，向控制台输出非常缓慢。将程序的输出发送到文件中并查看其速度是否更快可能会很值得尝试。

- Tom Tanner

Answer 4

我认为Oli Charlesworth准确地指出了超线程问题。我曾认为超线程就像真正拥有两个核心一样。但实际上并不是这样。我将其改为仅使用两个线程，结果达到了22,227,421，速度接近快了两倍。

- Dmobb Jr.

---

超线程只有在读写文件（或sleep调用）创建了大量锁定时才真正有用，这里处理器周期被浪费。因此，当您拥有一个永不退出的优化循环时，就像这样，超线程只是浪费了设置上下文切换的周期。 - Richard J. Ross III

你的意思是由于读/写内存而导致阻塞。如果整个线程在操作系统中休眠，超线程对非超线程没有帮助。但是，如果处理器必须等待多个时钟周期才能从内存中获取某些内容（或类似操作），则超线程会有所帮助。 - Mats Petersson

Answer 5

虽然 @MatsPetersson 是正确的（至少对于基于POSIX的系统来说，stdout是一个共享资源），但他没有提供解决这个问题的方法，所以下面是如何消除这些讨厌的锁定操作。

POSIX C定义了一个函数putc_unlocked，它将执行与putc完全相同的操作，但不会进行锁定（令人惊讶）。通过使用这个函数，我们可以定义自己的函数，在多线程场景中打印整数时不会进行锁定，并且比cout或printf更快：

void printint_unlocked(FILE *fptr, int i) {
    static int digits[] = {
        1,
        10,
        100,
        1000,
        10000,
        100000,
        1000000,
        10000000,
        100000000,
        1000000000,
    };

    if (i < 0) {
        putc_unlocked('-', fptr);
        i = -i;
    }

    int ndigits = (int) log10(i);
    while (ndigits >= 0) {
        int digit = (i / (digits[ndigits])) % 10;

        putc_unlocked('0' + digit, fptr);

        --ndigits;
    }
}

请注意，使用此方法可能会出现竞争条件，导致输出中的数字发生冲突。如果您的算法最终没有任何冲突，您仍应该获得多线程代码的性能提升。

第三种选择（可能对您的用例过于复杂）是在另一个线程上创建事件队列，并从该线程执行所有打印操作，从而不会出现竞争条件和线程之间的锁定问题。

- Richard J. Ross III

---

这种方法并不能真正解决问题，因为您还需要确保您的 putc_unlocked 不会相互干扰，使用某种互斥锁等。[而且我认为我可以想出一种更简洁的方法将整数转换为字符串] - Mats Petersson

@RaymondChen，实际上你可以像调用其他没有线程安全的方法一样调用这个方法——只是在未加锁的情况下不能保证线程安全。 - Richard J. Ross III

当然，@RaymondChen，我在我的答案中已经概述了这种方法的问题。但是，随着素数之间的间隔增加（达到70,000,000），每次打印之间的时间将减少，随着时间的推移减少竞争条件。对于64位整数，这些间隔可能不够长，但它肯定会加快速度。 - Richard J. Ross III

@RaymondChen 当然，但是如果没有为每个线程单独创建输出文件或者完全删除打印操作，你的选择就不多了。 - Richard J. Ross III

如果两个线程同时调用printint_unlocked，那么会发生未定义的行为。如果两个线程从未同时调用printint_unlocked，则两个线程在原始代码中永远不会同时调用“cout”，因此锁定从未创建瓶颈。锁定“cout”导致瓶颈的事实表明，printint_unlocked将在多个线程上同时被调用。 - Raymond Chen