带标签的 mpi4py 发送/接收

Question

如何将进程的等级作为标记传递给 mpi4py.MPI.COMM_WORLD.Send() 函数并使用 mpi4py.MPI.COMM_WORLD.Recv() 正确接收它？

我指的是sending and receiving messages between two processes using Send and Recv functions的以下代码示例

#passRandomDraw.py
import numpy
from mpi4py import MPI
comm = MPI.COMM_WORLD
rank = comm.Get_rank()

randNum = numpy.zeros(1)

if rank == 1:
    randNum = numpy.random.random_sample(1)
    print "Process", rank, "drew the number", randNum[0]
    comm.Send(randNum, dest=0)

if rank == 0:
    print "Process", rank, "before receiving has the number", randNum[0]
    comm.Recv(randNum, source=1)
    print "Process", rank, "received the number", randNum[0]

我想将发送进程的等级作为标签传递，以便接收进程可以在有多个发送者的情况下识别它。我就是这样做的

#passRandomDraw.py
import numpy
from mpi4py import MPI
comm = MPI.COMM_WORLD
rank = comm.Get_rank()

randNum = numpy.zeros(1)
rnk = -1 # EDIT

if rank == 1:
    randNum = numpy.random.random_sample(1)
    print "Process", rank, "drew the number", randNum[0]
    comm.Send(randNum, dest=0, tag=rank) # EDIT

if rank == 0:
    print "Process", rank, "before receiving has the number", randNum[0]
    print "Sender rank:", rnk
    comm.Recv(randNum, 1, rnk) # EDIT
    print "Process", rank, "received the number", randNum[0]
    print "Sender rank:", rnk # EDIT

对于接收进程（rank=0），我希望 rnk 的值为 1，但它仍然是 -1。

有人可以告诉我我在这里做错了什么吗？谢谢！

Answer 1

函数Recv 将接收到的消息存储在一个变量中。您必须提供预期发件人的等级。因此，您始终知道发件人是谁。消息传递接口永远不需要识别某人，该信息始终是系统固有的。

如果您期望来自同一发件人的多条消息，您可以使用标签来区分它们。您需要自己提供这些标签，没有自然的方法可以获取这些标签。只需以某种方式标记消息，给它们编号。

如果您有标签，Recv 函数只会在收到具有合适源和标签的消息时返回。这是一个阻塞函数调用。

在您的情况下，tag=-1 等于通用常量MPI.ANY_TAG（通过print MPI.ANY_TAG 验证），因此Recv 将接受任何标签。但它绝不会覆盖其输入变量rnk。试试rnk = -2 # EDIT，你会看到的。

你可以用不同的方式编写你的代码，虽然这不会改变底层逻辑（即你作为程序员必须始终知道发送者）它只是隐藏它，使其隐含：

#passRandomDraw.py
import numpy
from mpi4py import MPI
comm = MPI.COMM_WORLD
rank = comm.Get_rank()

randNum = numpy.zeros(1)
rnk = -1 # EDIT

if rank == 1:
    randNum = numpy.random.random_sample(1)
    print "Process", rank, "drew the number", randNum[0]
    comm.Send(randNum, dest=0, tag=rank) # EDIT

if rank == 0:
    print "Process", rank, "before receiving has the number", randNum[0]
    print "Sender rank:", rnk
    status = MPI.Status()
    comm.Recv(randNum, source=MPI.ANY_SOURCE, tag=MPI.ANY_TAG, status=status) # EDIT
    rnk = status.Get_source()
    print "Process", rank, "received the number", randNum[0]
    print "Sender rank:", rnk # EDIT

Answer 2

以下示例演示了如何在 mpi4py 中使用带有排名和标签的 send 和 recv 函数。同样的方法应该适用于Send 和Recv 函数。 MPI.Status 对象用于获取每个接收到的消息的源和标签。当 mpi4py 文档不足时，咨询examples and tutorials written in C 往往会有所帮助。

from mpi4py import MPI

def enum(*sequential, **named):
    """Handy way to fake an enumerated type in Python
    http://stackoverflow.com/questions/36932/how-can-i-represent-an-enum-in-python
    """
    enums = dict(zip(sequential, range(len(sequential))), **named)
    return type('Enum', (), enums)

# Define MPI message tags
tags = enum('READY', 'DONE', 'EXIT', 'START')

# Initializations and preliminaries
comm = MPI.COMM_WORLD   # get MPI communicator object
size = comm.Get_size()  # total number of processes
rank = comm.Get_rank()  # rank of this process
name = MPI.Get_processor_name()
status = MPI.Status()   # get MPI status object

if rank == 0:
    # Master process executes code below
    tasks = range(2*size)
    task_index = 0
    num_workers = size - 1
    closed_workers = 0
    print("Master starting with {} workers".format(num_workers))
    while closed_workers < num_workers:
        data = comm.recv(source=MPI.ANY_SOURCE, tag=MPI.ANY_TAG, status=status)
        source = status.Get_source()
        tag = status.Get_tag()
        if tag == tags.READY:
            # Worker is ready, so send it a task
            if task_index < len(tasks):
                comm.send(tasks[task_index], dest=source, tag=tags.START)
                print("Sending task {} to worker {}".format(task_index, source))
                task_index += 1
            else:
                comm.send(None, dest=source, tag=tags.EXIT)
        elif tag == tags.DONE:
            results = data
            print("Got data from worker {}".format(source))
        elif tag == tags.EXIT:
            print("Worker {} exited.".format(source))
            closed_workers += 1

    print("Master finishing")
else:
    # Worker processes execute code below
    print("I am a worker with rank {} on {}.".format(rank, name))
    while True:
        comm.send(None, dest=0, tag=tags.READY)
        task = comm.recv(source=0, tag=MPI.ANY_SOURCE, status=status)
        tag = status.Get_tag()

        if tag == tags.START:
            # Do the work here
            result = task**2
            comm.send(result, dest=0, tag=tags.DONE)
        elif tag == tags.EXIT:
            break

    comm.send(None, dest=0, tag=tags.EXIT)