计时多处理脚本

Question

在使用多处理模块时，我偶然发现了一个奇怪的计时问题。考虑以下场景。我有这样的功能：import multiprocessing as mpdef workerfunc(x):    # timehook 3    # something with x    # timehook 4def outer():    # do something    mygen = ... (some generator expression)    pool = mp.Pool(processes=8)    # time hook 1    result = [pool.apply(workerfunc, args=(x,)) for x in mygen]    # time hook 2if __name__ == '__main__':    outer()我正在利用时间模块来随意了解我的函数运行的时间。我成功地创建了 8 个独立的进程，它们无误地终止。工人完成的最长时间约为 130 毫秒（在时间钩 3 和 4 之间测量）。我预计（因为它们并行运行）钩子 1 和钩子 2 之间的时间大致相同。令人惊讶的是，我得到了 600 毫秒的结果。我的机器有 32 个内核，应该能够轻松处理这个问题。谁能给我一个提示，这种时间差异是从哪里来的？谢谢！

Answer 1

由于您使用的是多处理而不是多线程，因此您的性能问题与 GIL（Python 的全局解释器锁）无关。

我找到了一个有趣的链接，用一个例子解释了这一点，你可以在这个答案的底部找到它。

GIL 不会阻止进程在机器的不同处理器上运行。它只允许一个线程在解释器中一次运行。

所以多处理而不是多线程将使您实现真正的并发。

让我们通过一些基准测试来理解这一切，因为只有这样才能让您相信上面所说的。是的，这应该是学习的方式——体验它而不是仅仅阅读或理解它。因为如果你经历过一些事情，再多的争论也无法说服你接受相反的想法。

import random

from threading import Thread

from multiprocessing import Process

size = 10000000   # Number of random numbers to add to list

threads = 2 # Number of threads to create

my_list = []

for i in xrange(0,threads):

    my_list.append([])

def func(count, mylist):

    for i in range(count):

        mylist.append(random.random())

def multithreaded():

    jobs = []

    for i in xrange(0, threads):

        thread = Thread(target=func,args=(size,my_list[i]))

        jobs.append(thread)

    # Start the threads

    for j in jobs:

        j.start() 

    # Ensure all of the threads have finished

    for j in jobs:

        j.join()

def simple():

    for i in xrange(0, threads):

        func(size,my_list[i])

def multiprocessed():

    processes = []

    for i in xrange(0, threads):

        p = Process(target=func,args=(size,my_list[i]))

        processes.append(p)

    # Start the processes

    for p in processes:

        p.start()

    # Ensure all processes have finished execution

    for p in processes:

        p.join()

if __name__ == "__main__":

    multithreaded()

    #simple()

    #multiprocessed()

附加信息

在这里您可以找到此信息的来源和更详细的技术说明（奖励：其中还引用了 Guido Van Rossum :)）

Answer 2

您正在使用pool.apply哪个正在阻塞。pool.apply_async改为使用，然后函数调用将全部并行运行，并且每个调用都会AsyncResult立即返回一个对象。您可以使用此对象来检查进程何时完成，然后也可以使用此对象检索结果。