Python 多处理 apply_async 永远不会在 Windows 7 上返回结果

Question

我正在尝试遵循一个非常简单的多处理示例：

import multiprocessing as mp

def cube(x):
    return x**3

pool    = mp.Pool(processes=2)
results = [pool.apply_async(cube, args=x) for x in range(1,7)]

但是，在我的 windows 机器上，我无法获得结果（在 ubuntu 12.04LTS 上它运行完美）。

如果我检查 results，我会看到以下内容：

[<multiprocessing.pool.ApplyResult object at 0x01FF0910>,
 <multiprocessing.pool.ApplyResult object at 0x01FF0950>,
 <multiprocessing.pool.ApplyResult object at 0x01FF0990>,
 <multiprocessing.pool.ApplyResult object at 0x01FF09D0>,
 <multiprocessing.pool.ApplyResult object at 0x01FF0A10>,
 <multiprocessing.pool.ApplyResult object at 0x01FF0A50>]

如果我运行results[0].ready()，我总是会得到False。

如果我运行results[0].get()，python 解释器会冻结，等待得到永远不会出现的结果。

这个例子很简单，所以我认为这是一个与操作系统相关的低级错误（我在 Windows 7 上）。但也许其他人有更好的主意？

Answer 1

这里有几个错误。首先，您必须在if __name__ == "__main__": 保护when running on Windows 中声明Pool。其次，您必须将args 关键字参数传递给一个序列，即使您只传递一个参数。所以把它们放在一起：

import multiprocessing as mp

def cube(x):
    return x**3

if __name__ == "__main__":
    pool    = mp.Pool(processes=2)
    results = [pool.apply_async(cube, args=(x,)) for x in range(1,7)]
    print([result.get() for result in results])

输出：

[1, 8, 27, 64, 125, 216]

编辑：

哦，正如 moarningsun 所说，交互式解释器中的multiprocessing does not work well：

注意

此包中的功能要求 __main__ 模块是 孩子们可以导入。这在编程指南中有介绍 但是值得在这里指出。这意味着一些例子， 例如multiprocessing.Pool 的例子在 交互式解释器。

因此，您需要将代码作为脚本实际执行才能正确测试。

Answer 2

我正在运行 python 3，IDE 是 anaconda (windows) 中的 spyder，所以这个技巧对我不起作用。我尝试了很多，但没有任何区别。我得到了问题的原因，并且与 dano 在他的笔记中列出的相同。但是经过一整天的搜索，我得到了一些解决方案，它帮助我在我的 Windows 机器上运行相同的代码。该网站帮助我获得了解决方案：

http://python.6.x6.nabble.com/Multiprocessing-Pool-woes-td5047050.html

由于我用的是python 3，所以我把程序改了一点：

from types import FunctionType
import marshal

def _applicable(*args, **kwargs):
  name = kwargs['__pw_name']
  code = marshal.loads(kwargs['__pw_code'])
  gbls = globals() #gbls = marshal.loads(kwargs['__pw_gbls'])
  defs = marshal.loads(kwargs['__pw_defs'])
  clsr = marshal.loads(kwargs['__pw_clsr'])
  fdct = marshal.loads(kwargs['__pw_fdct'])
  func = FunctionType(code, gbls, name, defs, clsr)
  func.fdct = fdct
  del kwargs['__pw_name']
  del kwargs['__pw_code']
  del kwargs['__pw_defs']
  del kwargs['__pw_clsr']
  del kwargs['__pw_fdct']
  return func(*args, **kwargs)

def make_applicable(f, *args, **kwargs):
  if not isinstance(f, FunctionType): raise ValueError('argument must be a function')
  kwargs['__pw_name'] = f.__name__  # edited
  kwargs['__pw_code'] = marshal.dumps(f.__code__)   # edited
  kwargs['__pw_defs'] = marshal.dumps(f.__defaults__)  # edited
  kwargs['__pw_clsr'] = marshal.dumps(f.__closure__)  # edited
  kwargs['__pw_fdct'] = marshal.dumps(f.__dict__)   # edited
  return _applicable, args, kwargs

def _mappable(x):
  x,name,code,defs,clsr,fdct = x
  code = marshal.loads(code)
  gbls = globals() #gbls = marshal.loads(gbls)
  defs = marshal.loads(defs)
  clsr = marshal.loads(clsr)
  fdct = marshal.loads(fdct)
  func = FunctionType(code, gbls, name, defs, clsr)
  func.fdct = fdct
  return func(x)

def make_mappable(f, iterable):
  if not isinstance(f, FunctionType): raise ValueError('argument must be a function')
  name = f.__name__    # edited
  code = marshal.dumps(f.__code__)   # edited
  defs = marshal.dumps(f.__defaults__)  # edited
  clsr = marshal.dumps(f.__closure__)  # edited
  fdct = marshal.dumps(f.__dict__)  # edited
  return _mappable, ((i,name,code,defs,clsr,fdct) for i in iterable)

这个函数之后，上面的问题代码也稍微改成这样：

from multiprocessing import Pool
from poolable import make_applicable, make_mappable

def cube(x):
  return x**3

if __name__ == "__main__":
  pool    = Pool(processes=2)
  results = [pool.apply_async(*make_applicable(cube,x)) for x in range(1,7)]
  print([result.get(timeout=10) for result in results])

我得到的输出是：

[1, 8, 27, 64, 125, 216]

我认为这篇文章可能对某些 Windows 用户有用。