python多处理队列错误

Question

我有这个python代码来读取一个文件，进行一些处理，并并行地编写结果：

def line_chunker(path):
    """
    Reads a file in chunks and yields each chunk.
    Each chunk is guaranteed to end at a carriage return (EOL).
    Each chunk is returned a single string.

    The number of chunks the file is split into is equal to the number of CPU cores
    available
    """
    size = os.path.getsize(path)
    cores = mp.cpu_count()
    chunksize = size/cores # gives truncated integer

    f = open(path)
    s = f.readline() # skip header
    while True:
        part = f.readlines(chunksize)
        if not part:
            f.close()
            break
        else:
            yield "".join(part)
    f.close()

def _validate(chunk, outq):
    """ Performs format validation on a given chunk of a csv file """
    rows = csv.reader(StringIO(chunk))
    vld = validation.RowValidator(rows)
    vld.check_rows()
    outq.put(vld.errors)

def _write(outq):
    """Writes lines in the outq to a text file """
    outfile = open("C:/testoutput.txt", "w")
    while True:
        result = outq.get()
        if result is None:
            outfile.close()
            break
        else:
            for line in result:
                outfile.write(line)
                outfile.write("\n")

def validate_pll(path):    
    """ Perform validation in parallel """

    pool = mp.Pool()
    outq = mp.Manager().Queue(maxsize = 8)

    writer = mp.Process(target = _write, args = (outq,))
    writer.start()
    for chunk in line_chunker(path):
        pool.apply_async(_validate, (chunk, outq))

    pool.close()
    pool.join()

它以块的形式读取文件，并为每个块启动一个进程来进行处理。处理的结果被放到一个队列中，由另一个进程监视。

代码运行，但完成后我得到一个奇怪的EOFError。

我怀疑这是因为我没有调用writer.join()，但是如果我添加这一行，如下所示：

def validate_pll(path):    
    """ Perform validation in parallel """

    pool = mp.Pool()
    outq = mp.Manager().Queue(maxsize = 8)

    writer = mp.Process(target = _write, args = (outq,))
    writer.start()
    for chunk in line_chunker(path):
        pool.apply_async(_validate, (chunk, outq))

    pool.close()
    pool.join()
    writer.join()

代码只是挂起。知道我做错了什么吗？

给出的错误消息是：

Process Process-10:
Traceback (most recent call last):
    File C\Anaconda\lib\multiprocessing\process.py, line 258, in _bootstrap
        self.run()
    File C\Anaconda\lib\multiprocessing\process.py line 114, in run
       self._target(*self._args, **self._kwargs)
    File C:\SVN\PortfolioInspector\trunk\parallel.py, line 114 in _write
       result = outq.get()
    File "(string)", line 2, in get
    File C\Anaconda\lib\multiprocessing\managers.py, line 759, in _callmethod
        kind, result = conn.recv()
EOFError

Answer 1

当主进程结束时，_writer进程仍在等待将条目写入outq。它通过打开到管理共享Manager的Queue进程的阻塞连接来等待条目。现在，在主进程完成其执行时，Manager进程将关闭，这将向_writer打开的连接发送EOF，您将看到该异常。

要解决这个问题，您需要告诉_writer在主进程结束之前关闭(扩展到Manager进程关闭)。实际上，您已经有了这样的机制，您只是没有使用它；将一个None发送到outq，_writer就会有序地关闭。在writer.join()之前打个电话，一切都会好起来的。