python - 在 irecv 中将对象作为缓冲区传递 - TypeError : expected a writeable buffer object

Question

我正在一个项目中使用 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)

我期待以下行为:

1. 等级 0 的节点将数据包作为对象发送到等级 1 的节点

2. 排名为 1 的节点开始非阻塞接收，当它完成接收时，当 test() 返回 True 时会发生什么，我可以打印data.msg.

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

类型错误:需要一个可写的缓冲区对象

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

Answer 1

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

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

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)

请注意，您的发件人代码缺少消息的完整内容。但无论您发送还是isend数据都无关紧要。

无论如何，您必须以某种方式确定接收缓冲区的足够缓冲区大小，这可能不可能真正干净地完成。如果缓冲区太小，您将收到 MPI.Exception。

您还可以使用低级接口(interface)。例如，您可以轻松发送 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 必须匹配才能理解。