MapReduce with MPI question

I am doing an exercise using MPI to count frequencies of words distributed in several different files following similar steps in this instruction.

But I met a problem with step 2. In my implementation, I first sent out locally counted word-count pairs into corresponding processors according to the words' hashcode. At the same time, each processor might need to receive word-count pairs sented from other processors. As a result, one processor will have to both send and receive.

For example:

In processor 1, there are word-count pairs:

cat: 3

dog: 4

fox: 2

In processor 2, there are word-count pairs:

deer: 2

cat: 2

fox: 1

red: 3

and so on. Suppose the word "cat" is computed to be sent to processor 2, and the words "deer", "fox", "red" are supposed to be sent to processor 1. So the pseudocode for each processor would be:

for each word-count pair:
compute the word hash code
MPI_Issend it to corresponding processor according to the hash code

while (true):
MPI_Recv word-count pair
combine the counts of the same word in this processor

MPI_Waitall
Do some other things

Note that the second loop will never stop, so I need to add a termination condition. My way of implementing this is to send a signal from the manager processor with rank 0 to each worker processor. So the second loop becomes:

while (true):
MPI_Recv word-count pair
determine sender from the receive status
if (sender != manager)
combine the counts of the same word in this processor
else
break

But this algorithm turned out to be not stable. Although it will get the intended result in most runs, it can enter into deadlock. My guess for the failure is that some workers may receive the termination signal from the manager too early before receiving all the word-count pairs from other worker processors, so the corresponding MPI_Issend from those workers would have to wait indefinitely, thus a deadlock.

I wish someone could give some suggestions of improvement on my implementation. Or if someone have better algorithms for this step, that would also be appreciated! Thank you for your attention!

UPDATE:

In order to prevent the manager from sending termination signals before any workers finish receiving word-count pairs, I summed up the total number of words from all processors involved in the shuffle stage, say $N$. Then, every time a processor receives a word-count pair, it should also send a signal to the manager. Accordingly, in the manager part, a loop is added to Recv the these signals, so that the manager will only send the termination signals when all the word-count pairs are received by workers. In this way, the manager signal will never interrupt with content sent from workers and thus the recv loop in the worker can terminate correctly. Below is the final modified pseudocode for this:

Manager Part:
for (i = 1 : N)
MPI_Recv signal with tag "recvd"
MPI_Send termination signal to all processors involved in the shuffle stage

Worker Part:
for each word-count pair:
compute the word hash code
MPI_Issend it to corresponding processor according to the hash code

while (true):
MPI_Recv word-count pair
**MPI_Send signal with tag "recvd" to notify the manager that a wcp is received**
if (sender != manager)
combine the counts of the same word in this processor
else
break

MPI_Waitall
Do some other things

Although this algorithm worked successfully, I'm not sure whether I'm on the right track. Am I over complexing this problem?

As Wolfgang observed, you are basically transposing a connectivity matrix: every process $i$ knows row $i$ that states which $j$ values you're sending to. What you're interested in is the columns: the description in column $j$ of everyone that sends to $j$. However, implementing this with an Allgather is overkill: you only need to know the sum of the entries in a column.

MPI has just the routine for this! It's called MPI_Reduce_scatter. Every process declares an array of values as long as the number of ranks; MPI then does a pointwise reduction on the components of this array (that's where you do the summing), and then scatters it, sending a single scalar to each process.

This routine has two major applications: the setup phase of the sparse matrix-vector product, and the execution of a 2D-distributed dense matrix-vector product. (Look for reduce-scatter in the index of my HPC book) And it sounds like it's just what you need: because of your hash code everyone knows who they are sending to, not who they are receiving from.

Small note: in your receive call I would use MPI_ANY_SOURCE to prevent any deadlock or parallel performance problems.

What you are trying to do is difficult -- namely, to receive messages without knowing how many messages and from which processor. The problem is that you can never be sure that you've received all of them, or whether there are more coming but some processor is just slow to send something.

The solution to this problem is that you first need to figure out who each processor is going to send something to. Think of this as a $P\times P$ matrix: in row $p$, you're going to put a one in column $q$ if processor $p$ wants to send something to processor $q$, and a zero otherwise. Processor $q$ would then know to expect a message from all processors $p$ for which there is a one in column $q$ of the matrix.

Of course, each processor only knows its own row. To know what's in a column implies transposing this matrix. Take a look, for example, at the compute_point_to_point_communication_pattern() function that does that (without building a matrix, of course, but conceptually that's what it is doing): https://github.com/dealii/dealii/blob/master/source/base/mpi.cc#L91 Its documentation is here: https://www.dealii.org/8.5.0/doxygen/deal.II/namespaceUtilities_1_1MPI.html#a89b9a3309dffffe1447758157a33dbb6

Once you have this information, you can loop over all senders and wait for the data they want to send to you. (A better pattern is to just query for any incoming message and process it, until you know that every one of the known senders has sent you their message.)

• thanks very much for your suggestion, it really gave me a new insight on how to determine the information of senders, which I had already finished but seeems not to be quite efficient. But my problem here is actually to figure out a way to determine how many messages to expect from other processors. I had thinked out a way to achieve this but I'm not quite satisfied. I added an update on this, could you please take a look at it and leave some comments. Thank you so much! – user123 Jul 21 '17 at 8:06
• @David: That's one way to do it, though in general there is no reason to use a master-slave approach like you do for cases where the workers can communicate directly with each other. The exchange I outline in my answer above is no more expensive than sending data to the master from both sender and receiver, and it doesn't scale in general because you make a single processor a bottleneck. If you can, you should always avoid this, and my algorithm does this. – Wolfgang Bangerth Jul 21 '17 at 20:37
• I'll also note that your comment as written in the original question at this moment can't be correct since you have a while(true) loop that has no break or other exit statement. – Wolfgang Bangerth Jul 21 '17 at 20:38