多处理 IOError：错误消息长度

Posted 2023-02-16

【中文标题】多处理 IOError：错误消息长度

【英文标题】：Multiprocessing IOError: bad message length

【发布时间】：2015-08-30 07:08:01

【问题描述】：

将大参数传递给map 函数时，我得到一个IOError: bad message length。我怎样才能避免这种情况？ 当我设置N=1500 或更大时发生错误。

代码是：

import numpy as np
import multiprocessing

def func(args):
    i=args[0]
    images=args[1]
    print i
    return 0

N=1500       #N=1000 works fine

images=[]
for i in np.arange(N):
    images.append(np.random.random_integers(1,100,size=(500,500)))

iter_args=[]
for i in range(0,1):
    iter_args.append([i,images])

pool=multiprocessing.Pool()
print pool
pool.map(func,iter_args)

在 multiprocessing 的文档中，有一个函数 recv_bytes 会引发 IOError。会不会是因为这个？ (https://python.readthedocs.org/en/v2.7.2/library/multiprocessing.html)

编辑 如果我使用 images 作为 numpy 数组而不是列表，我会得到一个不同的错误：SystemError: NULL result without error in PyObject_Call。 有点不同的代码：

import numpy as np
import multiprocessing

def func(args):
    i=args[0]
    images=args[1]
    print i
    return 0

N=1500       #N=1000 works fine

images=[]
for i in np.arange(N):
    images.append(np.random.random_integers(1,100,size=(500,500)))
images=np.array(images)                                            #new

iter_args=[]
for i in range(0,1):
    iter_args.append([i,images])

pool=multiprocessing.Pool()
print pool
pool.map(func,iter_args)

EDIT2我实际使用的函数是：

def func(args):
    i=args[0]
    images=args[1]
    image=np.mean(images,axis=0)
    np.savetxt("image%d.txt"%(i),image)
    return 0

另外，iter_args 不包含同一组图像：

iter_args=[]
for i in range(0,1):
    rand_ind=np.random.random_integers(0,N-1,N)
    iter_args.append([i,images[rand_ind]])

【问题讨论】：

您传递了 3GB 的图像。如果它们从文件系统开始，也许您可​​以简单地将文件名排队。如果您正在生成它们，您可以使用多处理 Array 类在共享内存中创建数据，因此队列上实际的数据量（例如有关共享内存的信息）会更小。

在 ubuntu IOError: bad message length 上对我来说同样的错误

func()是需要同时对所有1500张图片进行操作，还是可以一次对单张图片进行操作？

你想达到什么目的？您提供的代码似乎无缘无故地使用了多处理：它将启动一个 single 子进程并将 all 图像传递给该进程。你真的不想要很多子进程，每个进程一次处理一个图像吗？

我在我的答案中添加了一个解决方案（没有多处理）。它将在相当现代的硬件上快速处理 1500 张图像，无需多处理。

【参考方案1】：

这就是解决问题的方法：将图像声明为全局的。

import numpy as np
import multiprocessing


N=1500       #N=1000 works fine

images=[]
for i in np.arange(N):
    images.append(np.random.random_integers(1,100,size=(500,500)))

def func(args):
    i=args[0]
    images=images
    print i
    return 0

iter_args=[]
for i in range(0,1):
    iter_args.append([i])

pool=multiprocessing.Pool()
print pool
pool.map(func,iter_args)

【讨论】：

这会将所有图像复制到每个进程 afaik，使得这种方法在缩放方面真的很差

【参考方案2】：

当运行你的程序时，它实际上给了我一个明确的错误：

OSError: [Errno 12] Cannot allocate memory

正如其他用户所提到的，您的问题的解决方案是简单地添加内存（大量）或更改您的程序处理图像的方式。

它使用这么多内存的原因是因为您在模块级别为图像分配内存。因此，当多进程分叉您的进程时，它还会复制所有图像（根据Shared-memory objects in python multiprocessing，这不是免费的），这不是必需的，因为您还将图像作为参数提供给多进程模块也使用 ipc 复制的函数和泡菜，这仍然可能导致内存不足。尝试其他用户提出的解决方案之一。

【参考方案3】：

您正在创建一个池并将所有图像一次发送到 func()。如果您可以一次处理单个图像，请尝试这样的操作，对于我来说，使用 Python 2.7.10 在 35 秒内完成 N=10000：

import numpy as np
import multiprocessing

def func(args):
    i = args[0]
    img = args[1]
    print ":  ".format(i, img.shape, img.sum())
    return 0

N=10000

images = ((i, np.random.random_integers(1,100,size=(500,500))) for i in xrange(N))
pool=multiprocessing.Pool(4)
pool.imap(func, images)
pool.close()
pool.join()

这里的关键是使用迭代器，这样您就不必一次将所有数据保存在内存中。例如，我将图像从包含所有数据的数组转换为生成器表达式，以便仅在需要时创建图像。您可以修改它以从磁盘或其他加载图像。我还使用了 pool.imap 而不是 pool.map。

如果可以，请尝试在工作函数中加载图像数据。现在，您必须序列化所有数据并将其传送到另一个进程。如果您的图像数据较大，这可能是一个瓶颈。

[更新现在我们知道 func 必须同时处理所有图像]

您可以对图像进行迭代均值。这是一个不使用多处理的解决方案。要使用多处理，您可以将图像分成块，然后将这些块放到池中。

import numpy as np

N=10000
shape = (500,500)

def func(images):
    average = np.full(shape, 0)
    for i, img in images:
        average += img / N
    return average

images = ((i, np.full(shape,i)) for i in range(N))

print func(images)

【讨论】：

这里的关键点是您需要关闭并加入池（使用pool.close() 和pool.join()）。

【参考方案4】：

Python 可能会将您的数据加载到您的 RAM 内存中，并且您需要此内存可用。您检查过您的计算机内存使用情况吗？

正如 Patrick 提到的，您正在加载 3GB 的数据，请确保在达到 32 位内存限制时使用 64 位版本的 Python。这可能会导致您的进程崩溃：32 vs 64 bits Python

另一个改进是使用 python 3.4 而不是 2.7。 Python 3 实现似乎针对非常大的范围进行了优化，请参阅Python3 vs Python2 list/generator range performance