getting OSError: too many open files when running longish test suite on code that uses multiprocessing

Question

运行 python 2.7，开发环境是OS X但是生产环境是linux.

我有一些代码正在尝试通过多处理来加速，我让它运行良好并观察到所需的理论加速。然后我去了运行上面的测试套件，经过几次测试，在所有后续测试中开始出现上述 OS 错误。如果我运行从我开始收到错误的那一刻开始进行测试，其中一些通过，然后我再次收到该错误。这是相当合乎逻辑的，只是一个完整性检查。

为了找出问题所在，我将 __builtin__ 的 open 和 close 调用替换为打印的调用（遵循中的建议）

import __builtin__
import traceback
import sys
openfiles = set()
oldfile = __builtin__.file
class newfile(oldfile):
    def __init__(self, *args):
        self.x = args[0]
        print "### OPENING %s ###" % str(self.x)
        traceback.print_stack(limit=20)
        print
        sys.stdout.flush()
        oldfile.__init__(self, *args)
        openfiles.add(self)
    def close(self):
        print "### CLOSING %s ###" % str(self.x)
        oldfile.close(self)
        openfiles.remove(self)
oldopen = __builtin__.open
def newopen(*args):
    return newfile(*args)
__builtin__.file = newfile
__builtin__.open = newopen

除了成百上千行 ### OPENING /dev/null ###，我看到了什么。

当我对完成相同任务但没有多处理的代码执行相同的操作时，我没有得到这样的文件连接，因此按理说多处理在这里有问题。 traceback 调用支持这一点，这表明罪魁祸首在这里：

  File "/System/Library/Frameworks/Python.framework/Versions/2.7/lib/python2.7/multiprocessing/process.py", line 250, in _bootstrap
    sys.stdin = open(os.devnull)

在这里发布 multiprocessing::process.py::_bootstrap 函数的代码，以防万一它有帮助：

def _bootstrap(self):
    from . import util
    global _current_process

    try:
        self._children = set()
        self._counter = itertools.count(1)
        try:
            sys.stdin.close()
            sys.stdin = open(os.devnull)
        except (OSError, ValueError):
            pass
        _current_process = self
        util._finalizer_registry.clear()
        util._run_after_forkers()
        util.info('child process calling self.run()')
        try:
            self.run()
            exitcode = 0
        finally:
            util._exit_function()
    except SystemExit, e:
        if not e.args:
            exitcode = 1
        elif isinstance(e.args[0], int):
            exitcode = e.args[0]
        else:
            sys.stderr.write(str(e.args[0]) + '\n')
            sys.stderr.flush()
            exitcode = 1
    except:
        exitcode = 1
        import traceback
        sys.stderr.write('Process %s:\n' % self.name)
        sys.stderr.flush()
        traceback.print_exc()

    util.info('process exiting with exitcode %d' % exitcode)
    return exitcode

并且，为了它的价值，我正在使用如下所示的代码调用多处理：

num_cpus = multiprocessing.cpu_count()
pool = multiprocessing.Pool(processes=num_cpus)
num_per_job = len(input_data) / num_cpus + 1
chunks = [input_data[num_per_job*i:num_per_job*(i+1)] for i in range(num_cpus)]
# TODO: ^^^ make this a list of generators
data = pool.map(get_output_from_input, chunks)
return itertools.chain.from_iterable(data)

那么，问题是：这是 multiprocessing 中的错误，还是我做错了什么？我真的很欢迎有借口花下一周的时间深入研究 multiprocessing 代码并弄清楚它是如何工作的，但我很难说服上级认为这是对我时间的有效利用。感谢任何有经验的人提供帮助！

Answer 1

您需要关闭池以终止子进程并释放用于与它们通信的管道。用 contextlib.closing 做，这样你就不必担心跳过关闭的异常。 closing 将在 with 块的末尾关闭池，包括异常退出时。所以，你永远不需要自己关闭。

此外，Pool.map 将其请求分块，这样您就不必自己动手了。我删除了那段代码，但 get_output_from_input 签名可能不正确（每个输入项将调用一次，而不是输入项列表一次）因此您可能需要进行一些修复。

import contextlib
num_cpus = multiprocessing.cpu_count()
with contextlib.closing(multiprocessing.Pool(processes=num_cpus)) as pool:
    data = pool.map(get_output_from_input, input_data)
    return itertools.chain.from_iterable(data)

getting OSError: too many open files when running longish test suite on code that uses multiprocessing

getting OSError: too many open files when running longish test suite on code that uses multiprocessing

python

file-io

multiprocessing

contextmanager

python-multiprocessing