在 irecv 中将对象作为缓冲区传递 - TypeError：期望一个可写的缓冲区对象答案

【问题标题】：Passing an object as a buffer in irecv - TypeError: expected a writeable buffer object在 irecv 中将对象作为缓冲区传递 - TypeError：期望一个可写的缓冲区对象
【发布时间】：2019-11-27 16:00:55
【问题描述】：

我正在为一个项目使用 MPI。我需要以非阻塞模式将一个包从一个节点传输到另一个节点。我正在用一个包含与我的通信逻辑相关的信息的类来组织这个包。

我正在使用函数 irecv() 进行一些测试，以获取我的通信请求和缓冲区以及 test() 以验证是否有消息到达。

MPI 缺少 Python 文档，因此我正在查看 source code 以获取更多信息，我正在使用的函数在哪里。

源代码声明irecv如下：

def irecv(self, buf=None, int source=ANY_SOURCE, int tag=ANY_TAG):
        """Nonblocking receive"""
        cdef MPI_Comm comm = self.ob_mpi
        cdef Request request = <Request>Request.__new__(Request)
        request.ob_buf = PyMPI_irecv(buf, source, tag, comm, &request.ob_mpi)
        return request

我知道，如果我想将数据放入缓冲区，我需要设置可选参数“buf”，并将接收到的消息存储在哪里。

我尝试了以下测试来了解它是如何工作的：

from mpi4py import MPI
import time

comm = MPI.COMM_WORLD
rank = comm.Get_rank()

class Package(object):
    msg = [[0,1,0,1,0,1],
        [0,1,0,1,0,1],
        [0,1,0,1,0,1],
        [0,1,0,1,0,1],
        [0,1,0,1,0,1]]

    gotMessage = False

    destination = -1

if rank == 0:
    data = Package()
    comm.isend(data, dest=1, tag=11)
elif rank == 1:
    data = Package()
    req = comm.irecv(buf=data, source=0, tag=11)
    while not req.test():
        sleep(0.1)
    print(rank, data.msg)

我期待以下行为：

等级为 0 的节点将数据包作为对象发送到等级为 1 的节点
rank 1 的节点开始非阻塞接收，当它完成接收时，当test() 返回True 时会发生什么，我可以打印data.msg。

问题是，当我运行buf时出现以下错误：

TypeError: 期望一个可写的缓冲区对象

如何正确使用irecv() 来传输/接收对象？

【问题讨论】：

标签： python mpi mpi4py

【解决方案1】：

在mpi4py 中，在 MPI 之上有两种接口。一个来回通信缓冲区的低级接口（用大写字母表示，即Isend），以及一个通信python对象的高级接口（即isend）。

高级接口通过 pickle 序列化对象。对于非阻塞操作，这需要一个用户提供的缓冲区并且需要足够大。另一方面，test 函数返回一个 found, object 元组。因此，使用高级接口，您的代码接收器如下所示：

buf = bytearray(b" " * 256)
req = comm.irecv(buf=buf, source=0, tag=11)
while True:
    found, data = req.test()
    if found:
        break
    time.sleep(0.1)
print(1, data.msg)

请注意，您的发件人代码缺少消息的完成。但不管你是send 还是isend 数据。

在任何情况下，您都必须以某种方式为接收缓冲区确定足够的缓冲区大小，这可能是不可能做到真正干净的。如果缓冲区太小，您将收到MPI.Exception。

您也可以使用低级接口。例如，您可以轻松地发送 numpy 数组：

if rank == 0:
    data = np.array([1, 2, 3], dtype=float)
    comm.Send(data, dest=1, tag=11)
elif rank == 1:
    data = np.zeros(3, dtype=float)
    req = comm.Irecv(buf=data, source=0, tag=11)
    while True:
        found = req.Test()
        if found:
            break
        time.sleep(0.1)
    print(1, data)

shape 和 dtype 必须匹配才能理解它。

【讨论】：