mpi4py Send/Recv with tag

Question

How can I pass a the rank of a process as a tag to the mpi4py.MPI.COMM_WORLD.Send() function and correctly receive it with mpi4py.MPI.COMM_WORLD.Recv()?

I'm referring to the following code example for 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]

I want to pass the rank of the sending process as a tag so that the receiving process can identify it in case there are multiple senders. This is what I do

#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

I expect the value of rnk to be 1 for the receiving process (which has rank=0), but it is still -1.

Can someone tell me what I'm doing wrong here? Thanks!

Solution 1

The function Recv will store a received message inside a variable. You have to supply the rank of the expected sender. Thus you always know who the sender is. A message passing interface does never need to identify someone, that information is always intrinsic to the system.

If you expect multiple message from the same sender you can distinguish these using tags. You need to supply these tags yourself, there is no natural way to obtain these. Just label the messages somehow, number them.

If you have a tag, the Recv function will only return when a message has been received which has a fitting source and tag. This is a blocking function call.

In your case, tag=-1 is equal to the universal constant MPI.ANY_TAG (verify via print MPI.ANY_TAG) and thus the Recv will accept any tag. But it will in no way overwrite its input variable rnk. Try rnk = -2 # EDIT and you'll see.

You can write your code differently, though this will not change the underlying logic (i.e. you as a programmer must always know the sender) it just hides it, makes it implicit:

#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

Solution 2

The following example demonstrates how to use the send and recv functions in mpi4py with ranks and tags. The same method should apply to the Send and Recv functions. An MPI.Status object is used to obtain the source and the tag for each received message. When the mpi4py docs are insufficient, it is often helpful to consult 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)

Comments

[user1953384]

That's very detailed. Thanks! Some of these got clearer as I read more into the tutorial.