python - 将 sys.stdout.write 与 pool.map 一起用于多处理 - 共享 sys.stdout

Question

这可能是我所缺少的非常简单的东西。

为什么我不能使用 pool.map(sys.stdout.write, iterable) ？

我可以使用 pool.map(len, iterable) 使用 same iterable 但是当使用 sys.stdout.write 我得到以下异常:

TypeError: expected string or Unicode object, NoneType found

这是痕迹:

Traceback (most recent call last):
  File "/home/reut/python/print_mult.py", line 19, in <module>
    pool.map(sys.stdout.write, messages)
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 251, in map
    return self.map_async(func, iterable, chunksize).get()
  File "/usr/lib/python2.7/multiprocessing/pool.py", line 558, in get
    raise self._value
TypeError: expected string or Unicode object, NoneType found

完整代码:

#!/usr/bin/env python

import multiprocessing
import sys

# pool of 10 workers
pool = multiprocessing.Pool(10)
messages = ["message #%d\n" % i for i in range(100)]
print messages
pool.map(sys.stdout.write, messages) # doesn't work - error
# print pool.map(len, messages) # works

编辑 #1 - ThreadPool 工作:

当我使用 ThreadPool(来自 multiprocessing.pool)时，它可以工作，所以我想这与无法共享 sys 有关。 stdout 跨进程流。

编辑 #2 - 手动流程也适用:

from multiprocessing import Process
import sys

# pool of 10 workers
processes = []
for i in range(10):
    processes.append(Process(target=sys.stdout.write, args=("I am process %d" % i, )))

for p in processes:
    p.start()

for p in processes:
    p.join()

所以现在我很困惑，因为我知道常规进程和映射进程之间的区别在于它的 fork 点。我不确定它在这里有什么关系。我唯一能想到的是 map 在内部存储 target 并且无法像 Process 的手动构造函数那样与工作人员共享它。

Answer 1

真正的错误是隐藏的。您只能传递可从模块命名空间直接引用的函数。但是，在某些情况下，有一些方法可以绕过此限制。 Unix 有一个特殊的功能，即一个进程可以被 fork ，它的所有内存都可以复制。这就是实例方法可以“传递”给子进程的方式——实际上没有传递任何东西。在 Windows 平台上，进程不能 fork ，而必须生成。这意味着启动了一个新的解释器。为了让解释器运行给定的函数，它被发送要运行的函数的名称和它所在的模块。解释器在最终运行函数之前导入模块并查找函数。

对于作为池一部分的进程，该进程已经启动，因此它无法从 fork 中获益以接收要运行的适当函数/方法的副本。相反，它必须使用与生成新进程时相同的技术。这就是为什么您可以让您的第二次编辑工作，但不能让池工作。

解决问题的最简单方法是使 print 成为函数而不是语句。

from __future__ import print_function

import multiprocessing
import sys

if __name__ == '__main__':
    pool = multiprocessing.Pool(2)
    messages = ["message #%d\n" % i for i in range(5)]
    print(messages) # <- notice the brackets around the arguments to print
    pool.map(print, messages)

如果失败，您可以定义一个函数来为您打印，并将其用作 map 的函数。

import multiprocessing 
import sys

def stdout_write(arg):
    sys.stdout.write(arg)

def stdout_print(arg):
    print arg

if __name__ == '__main__':
    pool = multiprocessing.Pool(2)
    messages = ["message #%d\n" % i for i in range(5)]
    print messages
    pool.map(stdout_print, messages)