concurrent.futures代码中的死锁

Question

我一直在尝试使用一些代码并行化,concurrent.futures.ProcessPoolExecutor但一直没有出现奇怪的死锁ThreadPoolExecutor.一个最小的例子:

from concurrent import futures

def test():
    pass

with futures.ProcessPoolExecutor(4) as executor:
    for i in range(100):
        print('submitting {}'.format(i))
        executor.submit(test)

在蟒蛇3.2.2(在64位的Ubuntu),这似乎在提交的所有作业后,始终悬挂-这似乎发生每当提交的作业数量比工人的数量.如果我ProcessPoolExecutor用ThreadPoolExecutor它替换它完美无瑕.

作为调查的尝试,我给每个未来一个回调打印值i:

from concurrent import futures

def test():
    pass

with futures.ProcessPoolExecutor(4) as executor:
    for i in range(100):
        print('submitting {}'.format(i))
        future = executor.submit(test)

        def callback(f):
            print('callback {}'.format(i))
        future.add_done_callback(callback)

这让我更加困惑 - i打印出来callback的值是它被调用时的值,而不是它被定义的时间(所以我从来没有看到,callback 0但我得到了很多callback 99s).再次,ThreadPoolExecutor打印出预期值.

想知道这可能是一个bug,我尝试了python的最新开发版本.现在,代码至少似乎终止了,但我仍然得到错误的i打印值.

任何人都可以解释:

• ProcessPoolExecutor在python 3.2和当前开发版本之间发生了什么,显然修复了这个死锁

• 为什么i要打印'错误'的价值

编辑:正如jukiewicz在下面指出的,当然打印i将在调用回调时打印值,我不知道我在想什么...如果我传递一个可调用对象,其值为i其属性之一,按预期工作.

编辑:更多信息:所有回调都被执行,所以看起来它executor.shutdown(被调用executor.__exit__)无法告诉进程已经完成.这似乎完全被固定在当前的Python 3.3,但似乎已经有很多变化的multiprocessing和concurrent.futures,所以我不知道什么固定这一点.由于我不能使用3.3(它似乎与numpy的发行版或开发版本兼容),我尝试简单地将其多处理和并发包复制到我的3.2安装,这似乎工作正常.尽管如此,据我所知,似乎有点奇怪 - ProcessPoolExecutor在最新发布的版本中完全被破坏但没有其他人受到影响.

Answer 1

我修改了代码如下，解决了这两个问题。函数被定义为闭包，因此将使用每次callback更新的值。i至于死锁，这可能是在所有任务完成之前关闭执行器的原因。等待期货完成也可以解决这个问题。

from concurrent import futures

def test(i):
    return i

def callback(f):
    print('callback {}'.format(f.result()))


with futures.ProcessPoolExecutor(4) as executor:
    fs = []
    for i in range(100):
        print('submitting {}'.format(i))
        future = executor.submit(test, i)
        future.add_done_callback(callback)
        fs.append(future)

    for _ in futures.as_completed(fs): pass

更新：哦，抱歉，我还没有阅读你的更新，这似乎已经解决了。