Python 中的多处理池 - 仅使用单个 CPU

Question

原始问题

我正在尝试在 Python 中使用多处理池。这是我的代码：

def f(x):
    return x
    
def foo():
    p = multiprocessing.Pool()
    mapper = p.imap_unordered
    
    for x in xrange(1, 11):
        res = list(mapper(f,bar(x)))

当xrange 像xrange(1, 6) 一样小时，此代码会使用所有 CPU（我有 8 个 CPU）。但是，当我将范围增加到xrange(1, 10) 时。我观察到只有 1 个 CPU 以 100% 的速度运行，而其余的只是空闲。可能是什么原因？是不是因为当我增加范围时，操作系统会因为过热而关闭 CPU？

我该如何解决这个问题？

最小的、完整的、可验证的例子

为了复制我的问题，我创建了这个例子：Its a simple ngram generation from a string problem。

#!/usr/bin/python

import time
import itertools
import threading
import multiprocessing
import random
    
    
def f(x):
    return x

def ngrams(input_tmp, n):
    input = input_tmp.split()

    if n > len(input):
        n = len(input)
        
    output = []
    for i in range(len(input)-n+1):
        output.append(input[i:i+n])
    return output 
    
def foo():
    
    p = multiprocessing.Pool()
    mapper = p.imap_unordered
    
    num = 100000000 #100
    rand_list = random.sample(xrange(100000000), num)
    
    rand_str = ' '.join(str(i) for i in rand_list)

    for n in xrange(1, 100):
        res = list(mapper(f, ngrams(rand_str, n)))

        
if __name__ == '__main__':
    start = time.time()
    foo()
    print 'Total time taken: '+str(time.time() - start)

当num 很小（例如num = 10000）时，我发现所有 8 个 CPU 都被使用了。然而，当num 相当大时（例如num = 100000000）。仅使用了 2 个 CPU，其余 CPU 处于空闲状态。这是我的问题。

警告：当num 太大时，可能会导致系统/VM 崩溃。

Answer 1

首先，ngrams 本身需要很多时间。虽然发生这种情况，但它显然只是一个一个核心。但即使完成后（只需将ngrams 调用移到mapper 之外并在其前后抛出print 就很容易测试），您仍然仍然使用一个核心。我得到 1 个核心 100%，其他核心都在 2% 左右。

如果您在 Python 3.4 中尝试同样的事情，情况会有些不同——我仍然以 100% 获得 1 个核心，但其他核心为 15-25%。

那么，发生了什么？好吧，在multiprocessing 中，传递参数和返回值总是有一些开销。在您的情况下，该开销完全淹没了实际工作，即return x。

开销的工作原理如下：主进程必须腌制值，然后将它们放入队列，然后等待另一个队列中的值并取消腌制它们。每个子进程都在第一个队列上等待，解开值，不做任何工作，腌制值，然后将它们放在另一个队列中。必须同步对队列的访问（通过大多数非 Windows 平台上的 POSIX 信号量，我认为是 Windows 上的 NT 内核互斥体）。

据我所知，您的进程 99% 以上的时间都在等待队列或读取或写入队列。

这并不是太出乎意料的，因为您有大量数据要处理，而且除了对数据进行酸洗和解酸之外根本没有任何计算。

如果您在CPython 2.7 中查看SimpleQueue 的源代码，酸洗和解酸会在持有锁的情况下发生。因此，您的任何后台进程所做的几乎所有工作都发生在持有锁的情况下，这意味着它们最终都在单个内核上序列化。

但在CPython 3.4 中，酸洗和解酸发生在锁之外。显然，这足以使用 15-25% 的核心。 （我相信这个变化发生在 3.2 中，但我懒得去追踪它。）

尽管如此，即使在 3.4 上，您等待访问队列的时间也远远多于做任何事情，甚至是 multiprocessing 开销。这就是为什么内核只能达到 25% 的原因。

当然，您在开销上花费的时间比实际工作要多几个数量级，这使得这不是一个很好的测试，除非您尝试测试您可以从特定 @987654331 中获得的最大吞吐量@ 在你的机器上实现或其他东西。

一些观察：

• 在您的实际代码中，如果您能找到一种方法来批量处理较大的任务（明确地，仅依赖 chunksize=1000 或此处的类似内容无济于事），那可能会解决您的大部分问题。
• 如果您的巨型数组（或其他任何东西）从未真正改变过，您可以将它传递到池初始化程序中，而不是在每个任务中传递，这几乎可以消除问题。
• 如果确实发生了变化，但只是从主流程方面，则可能值得分享而不是传递数据。
• 如果您需要从子进程中对其进行变异，请查看是否有办法对数据进行分区，以便每个任务都可以拥有一个切片而不会发生争用。
• 即使您需要具有显式锁定的完全竞争共享内存，它可能仍然比传递这么大的东西要好。
• 可能值得从 PyPI 获取 3.2+ 版本的 multiprocessing 或第三方 multiprocessing 库之一（或升级到 Python 3.x），只是为了将酸洗移出锁。

Answer 2

问题在于您的 f() 函数（在单独的进程上运行的函数）没有做任何特别的事情，因此它不会给 CPU 增加负载。

另一方面，ngrams() 正在执行一些“繁重”计算，但您是在主进程上调用此函数，而不是在池中。

为了让事情更清楚，考虑一下这段代码......

for n in xrange(1, 100):
    res = list(mapper(f, ngrams(rand_str, n)))

...等价于：

for n in xrange(1, 100):
    arg = ngrams(rand_str, n)
    res = list(mapper(f, arg))

以下是在您的主进程上执行的 CPU 密集型操作：

num = 100000000
rand_list = random.sample(xrange(100000000), num)

您应该更改您的代码以便在池中调用 sample() 和 ngrams()，或者更改 f() 以便它执行 CPU 密集型操作，您会看到所有CPU。