两者之间存在一些根本区别 Mystical's answer 和您的代码。

差异#1

您的代码为每个CPU创建一个工作块,并让它运行到完成。这意味着一旦一个线程完成,CPU使用量将急剧下降,因为一个CPU将处于空闲状态,而其他线程将运行到完成状态。这是因为日程安排并不总是公平的。一个线程的进度和完成速度可能比其他线程快得多。

OpenMP解决方案通过声明 schedule(dynamic) 它告诉OpenMP在内部创建一个工作队列,所有线程都将从中消耗工作。当一块工作完成后,本来会退出代码的线程会消耗另一块工作并忙于处理它。

最终,这成为一种平衡行为,即挑选足够大的块。太大,CPU可能在任务结束时没有达到最大值。太小,可能会有很大的开销。

差异#2

您正在写入变量, primes 在所有线程之间共享的。 这有两个后果:

• 它需要同步以防止 data race .
• 它在现代CPU上进行缓存 非常 不高兴,因为在一个线程上的写入对另一个线程可见之前,需要进行缓存刷新。

OpenMP解决方案通过以下方式解决了这一问题 reducing 通过 operator+() ,单个值的结果 素数 每个线程都保持在最终结果中。这就是为什么 reduction(+ : primes) 做

了解了OpenMP是如何拆分、安排工作和合并结果的,我们可以修改您的代码,使其行为类似。

#include <iostream>
#include <thread>
#include <vector>
#include <utility>
#include <algorithm>
#include <functional>
#include <mutex>
#include <future>

using namespace std;

int prime(int a, int b)
{
    int primes = 0;
    for (a; a <= b; a++) {
        int i = 2;
        while (i <= a) {
            if (a % i == 0)
                break;
            i++;
        }
        if (i == a) {
            primes++;
        }
    }
    return primes;
}


int workConsumingPrime(vector<pair<int, int>>& workQueue, mutex& workMutex)
{
    int primes = 0;
    unique_lock<mutex> workLock(workMutex);
    while (!workQueue.empty()) {
        pair<int, int> work = workQueue.back();
        workQueue.pop_back();

        workLock.unlock(); //< Don't hold the mutex while we do our work.
        primes += prime(work.first, work.second);
        workLock.lock();
    }
    return primes;
}


int main()
{
    int nthreads = thread::hardware_concurrency();
    int limit = 1000000;

    // A place to put work to be consumed, and a synchronisation object to protect it.
    vector<pair<int, int>> workQueue;
    mutex workMutex;

    // Put all of the ranges into a queue for the threads to consume.
    int chunkSize = max(limit / (nthreads*16), 10); //< Handwaving came picking 16 and a good factor.
    for (int i = 0; i < limit; i += chunkSize) {
        workQueue.push_back(make_pair(i, min(limit, i + chunkSize)));
    }

    // Start the threads.
    vector<future<int>> futures;
    for (int i = 0; i < nthreads; ++i) {
        packaged_task<int()> task(bind(workConsumingPrime, ref(workQueue), ref(workMutex)));
        futures.push_back(task.get_future());
        thread(move(task)).detach();
    }

    cout << "Number of logical cores: " << nthreads << "\n";
    cout << "Calculating number of primes less than " << limit << "... \n";

    // Sum up all the results.
    int primes = 0;
    for (future<int>& f : futures) {
        primes += f.get();
    }

    cout << "There are " << primes << " prime numbers less than " << limit << ".\n";
}

这仍然不是OpenMP示例行为的完美再现。例如,这更接近OpenMP static 因为工作块是固定大小的。此外,OpenMP根本不使用工作队列。所以我可能撒了一点谎,称之为善意的谎,因为我想更明确地表明工作被分割了。它可能在幕后做的是存储下一个线程在可用时应该开始的迭代,以及下一个块大小的启发式。

即使有这些差异,我也能够在很长一段时间内最大限度地使用所有CPU。

展望未来。。。

您可能注意到OpenMP版本的可读性要高得多。这是因为它旨在解决这样的问题。因此,当我们试图在没有库或编译器扩展的情况下解决它们时,我们最终会重新发明轮子。幸运的是,为了将这种功能直接引入C++,我们做了很多工作。具体来说 Parallelism TS 如果我们可以将其表示为标准C++算法,可以帮助我们解决这个问题。然后,我们可以告诉库将算法分发到所有CPU,因为它认为合适,所以它可以为我们完成所有繁重的工作。

在C++11中,在Boost的帮助下,此算法可以写成:

#include <iostream>
#include <iterator>
#include <algorithm>
#include <boost/range/irange.hpp>

using namespace std;

bool isPrime(int n)
{
    if (n < 2)
        return false;

    for (int i = 2; i < n; ++i) {
        if (n % i == 0)
            return false;
    }
    return true;
}

int main()
{
    auto range = boost::irange(0, 1000001);
    auto numPrimes = count_if(begin(range), end(range), isPrime);
    cout << "There are " << numPrimes << " prime numbers less than " << range.back() << ".\n";
}

为了将算法并行化,您只需要 #include <execution_policy> 然后通过 std::par 作为第一个参数 count_if .

auto numPrimes = count_if(par, begin(range), end(range), isPrime);

这就是让我乐于阅读的代码。

注:绝对没有花时间优化此算法。如果我们要做任何优化,我会研究类似于 Sieve of Eratosthenes 它使用以前的素数计算来帮助将来的计算。

首先,您需要意识到OpenMP通常有一个相当复杂的线程池,因此(精确地)匹配它可能至少有些困难。

其次,在我看来,在优化线程之前 应该 尝试至少从一个中等水平的基本算法开始。在这种情况下,您正在实现的基本算法基本上相当糟糕。它检查数字是否是质数,但做了很多工作,却没有完成任何有用的事情。

1. 它检查偶数是否为素数。除了2个,其他都不是。曾经
2. 它检查奇数是否可以被偶数整除。再说一遍,他们不是。曾经
3. 它检查数字是否可以被大于平方根的数字整除。如果没有比平方根小的除数,也不可能有比平方根大的除数。

虽然它可能不会影响速度,但我也发现使用一个函数来检查 仅有一个的 数字是质数,只返回 true / false 为了指示结果,而不是有一些复杂的代码来判断前一个循环是完成了还是提前退出了。

你 可以 通过消除更多的因素来优化算法,但这并没有给我带来“优化”的印象,我只是简单地避免了完全不必要的悲观。

至少在我看来,它(在这种情况下)也更容易使用 std::async 启动线程。这让我们可以很容易地从线程返回一个值(我们想要返回的计数)。

那么,让我们从修复开始 prime 根据这些观察结果:

int prime(int a, int b)
{
    int count = 0;

    if (a == 2)
        ++count;

    if (a % 2 == 0)
        ++a;

    auto check = [](int i) -> bool {
        for (int j = 3; j*j <= i; j += 2)
            if (i % j == 0)
                return false;
        return true;
    };

    for (a; a <= b; a+=2) {
        if (check(a))
            ++count;
    }
    return count;
}

现在,让我指出,这已经足够快了(即使是单线程),如果我们只想让工作完成得比完美的线程伸缩快4倍左右,我们就完成了,即使根本不使用线程。就你给出的极限而言,这在1秒钟内完成。

然而,为了便于讨论,让我们假设我们希望获得更多,并且也使用多个内核。这里需要认识到的一点是,我们通常希望至少比内核多几个线程。问题相当简单:每个核只有一个线程,我们没有什么可以弥补的事实是,我们甚至没有在线程之间真正分配负载——处理最大数量的线程比处理最小数量的线程要做的工作多得多——但如果我们有一台(例如)4核机器,只要一个线程完成,我们只能使用75%的CPU。然后当另一条线结束时,它会下降到50%。然后25%,最后完成,只使用一个核心。

我们 能够 可能会进行一些计算以尝试更均匀地分配负载,但只需将负载拆分为核心线程数的六到八倍就容易多了。这样,计算可以继续使用所有内核,直到只剩下三个线程为止 ^1. .

将所有这些放入代码中,我们可以得到如下结果:

int main() {
    using namespace chrono;

    int limit = 50000000;
    unsigned int nthreads = 8 * thread::hardware_concurrency();

    cout << "\nComputing multi-threaded:\n";
    cout << "Number of threads: " << nthreads << "\n";
    cout << "Calculating number of primes less than " << limit << "... \n";

    auto start2 = high_resolution_clock::now();

    vector<future<int>> threads;

    int intrvl = limit / nthreads;

    for (int i = 0; i < nthreads; i++)
        threads.emplace_back(std::async(std::launch::async, prime, i*intrvl + 1, (i + 1)*intrvl));

    int primes = 0;
    for (auto &t : threads)
        primes += t.get();

    auto end2 = high_resolution_clock::now();

    cout << "Primes: " << primes << ", Time: " << duration_cast<milliseconds>(end2 - start2).count() << "\n";
}

注意几点:

1. 这运行得足够快,我已经将上限提高了相当大的系数,所以它运行的时间足够长,我们至少可以看到它在运行前几秒钟使用了100%的CPU时间 ^2. .
2. 我添加了一些计时代码,以便更准确地了解它的运行时间。

至少当我运行这个程序时,它的行为似乎与我们预期/希望的一样:它使用100%的CPU时间,直到它接近尾声,当它在完成之前开始下降时(即,当我们要执行的线程少于要执行的内核时)。

1. 如果您想知道OpenMP是如何避免这种情况的:它通常使用线程池,因此循环的一些迭代次数作为一项任务分派给线程池。这使得它可以生成大量任务,而不会同时有大量线程争用CPU时间。
2. 根据您使用的上限,它在我的机器上大约90毫秒内完成,这段时间还不够长,甚至不能在CPU使用率图上显示一个明显的点。

OpenMP示例对sum变量使用“reduction” primes 这意味着每个任务对其本地 素数 变量 OpenMP添加线程本地副本 素数 在平行部分的末尾加起来就可以得到总数。 这意味着它不需要锁定。 正如@Sam所说,如果线程无法获取互斥锁,它就会进入休眠状态。 因此,在您的情况下,线程将花费相当长的睡眠时间。 如果不想使用OpenMP,请尝试 static std::atomic<int> primes = 0; 那么就不需要互斥锁的锁定和解锁。

或者可以使用数组模拟OpenMP缩减 primes[numThreads] where线程 i 总计到 primes[i] 然后求和 primes[] 最后。