Python 多处理 apply_async "assert left > 0" AssertionError

Question

我正在尝试在池中异步加载 numpy 文件：

self.pool = Pool(2, maxtasksperchild = 1)
...
nextPackage = self.pool.apply_async(loadPackages, (...))
for fi in np.arange(len(files)):
    packages = nextPackage.get(timeout=30)
    # preload the next package asynchronously. It will be available
    # by the time it is required.
    nextPackage = self.pool.apply_async(loadPackages, (...))

方法“loadPackages”：

def loadPackages(... (2 strings & 2 ints) ...):
    print("This isn't printed!')
    packages = {
        "TRUE": np.load(gzip.GzipFile(path1, "r")),
        "FALSE": np.load(gzip.GzipFile(path2, "r"))
    }
    return packages

甚至在加载第一个“包”之前，就会出现以下错误：

线程 Thread-8 中的异常：回溯（最近一次调用最后一次）：
文件“C:\Users\roman\Anaconda3\envs\tsc1\lib\threading.py”，第 914 行， 在 _bootstrap_inner self.run() 文件“C:\Users\roman\Anaconda3\envs\tsc1\lib\threading.py”，第 862 行，在 跑 self._target(*self._args, **self._kwargs) 文件“C:\Users\roman\Anaconda3\envs\tsc1\lib\multiprocessing\pool.py”，行 463，在_handle_results中 任务 = get() 文件 "C:\Users\roman\Anaconda3\envs\tsc1\lib\multiprocessing\connection.py", 第 250 行，在接收 buf = self._recv_bytes() 文件“C:\Users\roman\Anaconda3\envs\tsc1\lib\multiprocessing\connection.py”， 第 318 行，在 _recv_bytes 中 返回 self._get_more_data(ov, maxsize) 文件 "C:\Users\roman\Anaconda3\envs\tsc1\lib\multiprocessing\connection.py", 第 337 行，在 _get_more_data 中 向左断言 > 0 AssertionError

我密切监视资源：内存不是问题，发生错误时我还有很多剩余。 解压缩的文件只是普通的多维 numpy 数组。 单独使用具有更简单方法的 Pool 是可行的，并且像这样加载文件是可行的。只有结合起来才会失败。 （所有这些都发生在自定义的 keras 生成器中。我怀疑这是否有帮助，但谁知道呢。）Python 3.5。

此问题的原因可能是什么？如何解释这个错误？

感谢您的帮助！

Answer 1

Python C 核心代码中存在一个错误，该错误会阻止大于 2GB 的数据响应正确返回到主线程。 您需要按照上一个答案中的建议将数据拆分成更小的块，或者不为此功能使用多处理

我将此错误报告给 python 错误列表 (https://bugs.python.org/issue34563) 并创建了一个 PR (https://github.com/python/cpython/pull/9027) 来修复它，但它可能需要一段时间才能发布（更新：该修复程序存在于 python 3.8.0+)

如果您有兴趣，可以在我发布的链接中的错误描述中找到有关导致错误的原因的更多详细信息

Answer 2

它认为我找到了一种解决方法，方法是检索小块数据。在我的例子中，它是一个列表列表。

我有：

for i in range(0, NUMBER_OF_THREADS):
    print('MAIN: Getting data from process ' + str(i) + ' proxy...')
    X_train.extend(ListasX[i]._getvalue())
    Y_train.extend(ListasY[i]._getvalue())
    ListasX[i] = None
    ListasY[i] = None
    gc.collect()

改为：

CHUNK_SIZE = 1024
for i in range(0, NUMBER_OF_THREADS):
    print('MAIN: Getting data from process ' + str(i) + ' proxy...')
    for k in range(0, len(ListasX[i]), CHUNK_SIZE):
        X_train.extend(ListasX[i][k:k+CHUNK_SIZE])
        Y_train.extend(ListasY[i][k:k+CHUNK_SIZE])
    ListasX[i] = None
    ListasY[i] = None
    gc.collect()

现在它似乎可以工作了，可能是一次序列化更少的数据。 因此，也许如果您可以将数据分割成更小的部分，您就可以克服这个问题。祝你好运！