IPC在减速吗？

Question

我知道在使用多处理模块时会有开销，但这似乎是一个很高的开销，从我收集的数据来看，IPC的级别应该是相当低的。你知道吗

假设我生成一个1-1000之间的随机数的大列表，并且只想得到一个素数列表。此代码仅用于测试CPU密集型任务上的多处理。忽略首要性测试的总体无效性。你知道吗

大部分代码可能如下所示：

from random import SystemRandom
from math import sqrt
from timeit import default_timer as time
from multiprocessing import Pool, Process, Manager, cpu_count

rdev = SystemRandom()
NUM_CNT = 0x5000
nums = [rdev.randint(0, 1000) for _ in range(NUM_CNT)]
primes = []


def chunk(l, n):
    i = int(len(l)/float(n))
    for j in range(0, n-1):
        yield l[j*i:j*i+i]
    yield l[n*i-i:]


def is_prime(n):
    if n <= 2: return True
    if not n % 2: return False
    for i in range(3, int(sqrt(n)) + 1, 2):
        if n % i == 0:
            return False
    return True

在我看来，我应该能够把它分成多个进程。我有8个逻辑核，所以我应该能够使用cpu_count()作为进程的#。你知道吗

序列号：

def serial():
    global primes
    primes = []
    for num in nums:
        if is_prime(num):
            primes.append(num)  # primes now contain all the values

以下尺寸的NUM_CNT对应于速度：

• 0x500=0.00100秒。你知道吗
• 0x5000=0.01723秒。你知道吗
• 0x50000=0.27573秒。你知道吗
• 0x500000=4.31746秒。你知道吗

这就是我选择的多重处理方式。它使用chunk()函数将nums分成cpu_count()（大致相等）部分。它将每个块传递到一个新的进程中，该进程对它们进行迭代，然后将其分配给一个共享dict变量的条目。当我将值赋给共享变量时，IPC应该真正发生。否则为什么会发生？你知道吗

def loop(ret, id, numbers):
    l_primes = []
    for num in numbers:
        if is_prime(num):
            l_primes.append(num)
    ret[id] = l_primes


def parallel():
    man = Manager()
    ret = man.dict()
    num_procs = cpu_count()
    procs = []
    for i, l in enumerate(chunk(nums, num_procs)):
        p = Process(target=loop, args=(ret, i, l))
        p.daemon = True
        p.start()
        procs.append(p)
    [proc.join() for proc in procs]
    return sum(ret.values(), [])

同样，我预计会有一些开销，但时间似乎比串行版本增长的速度要快得多。你知道吗

• 0x500=0.37199秒。你知道吗
• 0x5000=0.91906秒。你知道吗
• 0x50000=8.38845秒。你知道吗
• 0x500000=119.37617秒。你知道吗

是什么导致它这样做的？是IPC吗？最初的设置让我期待一些开销，但这只是一个疯狂的数额。你知道吗

编辑：

下面是我如何计时函数的执行：

if __name__ == '__main__':
    print(hex(NUM_CNT))
    for func in (serial, parallel):
        t1 = time()
        vals = func()
        t2 = time()
        if vals is None:  # serial has no return value
            print(len(primes))
        else:  # but parallel does
            print(len(vals))
        print("Took {:.05f} sec.".format(t2 - t1))

每次使用相同的数字列表。你知道吗

输出示例：

0x5000
3442
Took 0.01828 sec.
3442
Took 0.93016 sec.

Answer 1

嗯。你怎么测量时间？在我的电脑上，并行版本比串行版本快得多。你知道吗

我是这样使用time.time()：如果我们假设tt是time.time()的别名。你知道吗

serial()
t2 = int(round(tt() * 1000))
print(t2 - t1)
parallel()
t3 = int(round(tt() * 1000))
print(t3-t2)

我得到，0x500000作为输入：

• 串行版本为5519ms
• 并行版本为3351ms

我认为你的错误是因为在并行程序中包含了数字生成过程，而不是在串行程序中。你知道吗

在我的电脑上，随机数的产生大约需要45秒（这是一个非常缓慢的过程）。所以，它可以解释你的两个值之间的差异，因为我不认为我的电脑使用了一个非常不同的架构。你知道吗