Python multiprocessing.Processgetting 在尝试返回结果时生成 AssertionError？答案

【问题标题】：Python multiprocessing.Processgetting generates AssertionError when attempting to get results returned?Python multiprocessing.Processgetting 在尝试返回结果时生成 AssertionError？
【发布时间】：2019-01-24 13:25:44
【问题描述】：

我正在尝试将multiprocessing 用于作为单个进程完成时非常慢的任务。正如您在下面的代码中看到的，每个进程都应该返回一些结果（return_dict）。我最初使用 10K 行数据集（数据存储在docs.txt 文件中，大约 70mb）测试了这段代码，并且代码按预期运行。但是，当我将脚本用于完整数据集（大约 5.6gb）时，我得到了一个AssertionError，如我的问题底部所示。我想知道是否有人知道可能导致它的原因以及我如何能够避免它。谢谢。

from multiprocessing import Process, Manager
import os, io, numpy
from gensim.models.doc2vec import Doc2Vec

def worker(i, data, return_dict):
    model = Doc2Vec.load("D:\\Project1\\doc2vec_model_DM_20180814.model")
    results = numpy.zeros((len(data), model.vector_size))
    for id, doc in enumerate(data):
        results[id,:] = model.infer_vector(doc, alpha = 0.01, steps = 100)
    return_dict[i] = results

if __name__ == '__main__':
    import time
    a  = time.time()
    path = "D:\\Project1\\docs.txt"    # <<=== data stored in this file
    data = []
    manager = Manager()
    jobs = []
    return_dict = manager.dict()

    with io.open(path, "r+", encoding = "utf-8") as datafile:
        for id, row in enumerate(datafile):
            row = row.strip().split('\t')[0].split()
            data.append(row)

    step = numpy.floor(len(data)/20)
    intervals = numpy.arange(0, len(data), step = int(step)).tolist()
    intervals.append(len(data))

    for i in range(len(intervals) - 1):
        p = Process(target=worker, args=(i, data[intervals[i]:intervals[i+1]], return_dict))
        jobs.append(p)
        p.start()
    for proc in jobs:
        proc.join()

    results = numpy.zeros((len(data), 1000))
    start = 0
    end = 0
    for _, result in return_dict.items():    #<<===Where error happens
        end = end + result.shape[0]
        results[start:end,:] = result[:,:]
        start = end

    print(time.time() - a)

错误信息：

Traceback (most recent call last):
  File "D:\Project1\multiprocessing_test.py", line 43, in <module>
    for _, result in return_dict.items():
  File "<string>", line 2, in items
  File "C:\ProgramData\Anaconda3\lib\multiprocessing\managers.py", line 757, in _callmethod
    kind, result = conn.recv()
  File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 250, in recv
    buf = self._recv_bytes()
  File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 318, in _recv_bytes
    return self._get_more_data(ov, maxsize)
  File "C:\ProgramData\Anaconda3\lib\multiprocessing\connection.py", line 337, in _get_more_data
    assert left > 0
AssertionError

【问题讨论】：

整个过程你的内存使用情况如何？内存使用量应在2*5.6gb 和3*5.6gb 之间，包括虚拟内存。
内存可能是个问题？ result_dict 包含 20 个元素，每个元素是一个 numpy 数组，大小约为 425k 行和 1000 列浮点值。我的电脑有 256 GB 内存，在实际的多处理阶段，电脑有 50% 的内存可用。错误发生时我在我的电脑旁，所以我不知道当时的内存使用情况如何。
好吧，既然你有这么大的内存，我想这不会是内存问题。基本上，您的子进程并没有真正将整个数据传输到您的主进程，这似乎是一个问题。这很奇怪，因为我认为 return_dict[i] = results 应该是一个阻塞操作。我无法调试您的代码，因此无法找出根本原因。另一种解决方案可以使用concurrent.futures.ProcessPoolExecutor，因为它允许您获取目标函数的返回值：future = executor.submit(args) & result = future.result()
Python multiprocessing apply_async "assert left > 0" AssertionError的可能重复

标签： python python-3.x multiprocessing python-multithreading

【解决方案1】：

这是我的情况和解决方案。希望这能有所帮助！我有一个名为“func”的函数来处理

partial_func = partial(func,a=params1,b=params2)
for i, _ in enumerate(pool.imap(partial_func, [1]))):
    pass

根本原因是我传递给“partial_func”的 params1 和 params2 太大。

【讨论】：

【解决方案2】：

我猜你正在使用所有可用内存。 dict.items() 创建一个包含所有项目并使用大量内存的 dict 副本。最好只使用 dict.iteritems() 来迭代您的结果。

编辑：抱歉，我一开始没有注意到 python-3 标签。在 Python3 中，dict.items()不再返回副本，应该可以使用了。

多处理中connection.py的相关代码为

left = _winapi.PeekNamedPipe(self._handle)[1]
assert left > 0

你在用windows吗？所以我猜这是一些与windows相关的问题，似乎PeekNamedPipe返回0。

【讨论】：

是的，你可以只使用 items()。 Python 3 items() 是 Python 2 iteritems()。