Collective Communication Patterns¶

With independent, often distributed processes, there is a need in many program situations to have all the processes communicating with each other, usually by sharing data, either before or after independent simultaneous computations that each process performs. Here we see simple examples of these collective communication patterns.

12. Collective Communication: Reduction on individual value¶

Once processes have performed independent concurrent computations, possibly on some portion of decomposed data, it is quite common to then reduce those individual computations into one value. This example shows a simple calculation done by each process being reduced to a sum and a maximum. In this example, MPI, has built-in computations, indicated by MPI_SUM and MPI_MAX in the following code. With four processes whose computed value for the variable square is shown in each oval, the code is implemented like this:

To do:

Try varying the number of processes using the -np flag. Do the results match with your understanding of how the reduction pattern works for multiple processes?

13. Collective Communication: Reduction on array of values¶

/* reduction2.c
  * ... illustrates the use of MPI_Reduce() using arrays...
  */

#include <mpi.h>
  #include <stdio.h>

#define ARRAY_SIZE 5

void printArray(int id, char* arrayName, int* array, int SIZE);
  void printSeparator(char* separator, int id);

int main(int argc, char** argv) {
      int myRank = -1;
      int srcArr[ARRAY_SIZE] = {0};
      int destArr[ARRAY_SIZE] = {0};

MPI_Init(&argc, &argv);
      MPI_Comm_rank(MPI_COMM_WORLD, &myRank);

if (myRank == 0) {
          printf("\nBefore reduction: ");
          printArray(myRank, "destArr", destArr, ARRAY_SIZE);
      }

for (unsigned i = 0; i < ARRAY_SIZE; i++) {
          srcArr[i] = myRank * i;
      }

printSeparator("", myRank);
      printArray(myRank, "srcArr", srcArr, ARRAY_SIZE);
      printSeparator("----", myRank);

MPI_Reduce(srcArr, destArr, ARRAY_SIZE, MPI_INT, MPI_SUM, 0, MPI_COMM_WORLD);

if (myRank == 0) {
          printf("\nAfter reduction:  ");
          printArray(myRank, "destArr", destArr, ARRAY_SIZE);
          printf("\n");
      }

MPI_Finalize();

return 0;
  }

/* utility to display an array
  * params: id, the rank of the current process
  *         arrayName, the name of the array being displayed
  *         array, the array being displayed
  *         SIZE, the number of items in array.
  * postcondition:
  *         the id, name, and items in array have been printed to stdout.
  */
  void printArray(int id, char* arrayName, int * array, int SIZE) {
      printf("Process %d, %s: [", id, arrayName);
      for (int i = 0; i < SIZE; i++) {
          printf("%3d", array[i]);
          if (i < SIZE-1) printf(",");
      }
      printf("]\n");
  }

/* utility to print a separator string between before and after sections.
  * params: separator, a string
  *         id, the rank of the current process.
  * postcondition: the conductor process has printed the separator.
  */
  void printSeparator(char* separator, int id) {
      MPI_Barrier(MPI_COMM_WORLD);
      if (id == 0) { printf("%s", separator); }
      MPI_Barrier(MPI_COMM_WORLD);
  }

To do:

Can you explain the reduction, MPI_reduce, in terms of srcArr and destArr?

Further Exploration:

This useful MPI tutorial explains other reduction operations that can be performed. You could use the above code or the previous examples to experiment with some of these.

14. Collective communication: Scatter for data decomposition¶

If processes can independently work on portions of a larger data array using the geometric data decomposition pattern, the scatter pattern can be used to ensure that each process receives a copy of its portion of the array. Process 0 gets the first chunk, process 1 gets the second chunk and so on until the entire array has been distributed.

/* scatter.c
  * ... illustrates the use of MPI_Scatter()...
  */

#include <mpi.h>      // MPI
  #include <stdio.h>    // printf(), etc.
  #include <stdlib.h>   // malloc()

void print(int id, char* arrName, int* arr, int arrSize);

int main(int argc, char** argv) {
      const int MAX = 8;
      int* arrSend = NULL;
      int* arrRcv = NULL;
      int numProcs = -1, myRank = -1, numSent = -1;

MPI_Init(&argc, &argv);                            // initialize
      MPI_Comm_size(MPI_COMM_WORLD, &numProcs);
      MPI_Comm_rank(MPI_COMM_WORLD, &myRank);

if (myRank == 0) {                                 // conductor process:
          arrSend = (int*) malloc( MAX * sizeof(int) );  //  allocate array1
          for (int i = 0; i < MAX; i++) {                //  load with values
              arrSend[i] = (i+1) * 11;
          }
          print(myRank, "arrSend", arrSend, MAX);        //  display array1
      }

numSent = MAX / numProcs;                          // all processes:
      arrRcv = (int*) malloc( numSent * sizeof(int) );   //  allocate array2

MPI_Scatter(arrSend, numSent, MPI_INT, arrRcv,     //  scatter array1
                  numSent, MPI_INT, 0, MPI_COMM_WORLD); //   into array2

print(myRank, "arrRcv", arrRcv, numSent);          // display array2

free(arrSend);                                     // clean up
      free(arrRcv);
      MPI_Finalize();
      return 0;
  }

void print(int id, char* arrName, int* arr, int arrSize) {
      printf("Process %d, %s: ", id, arrName);
      for (int i = 0; i < arrSize; i++) {
          printf(" %d", arr[i]);
      }
      printf("\n");
  }

To do:

Find documentation for the MPI function MPI_Scatter. Make sure that you know what each parameter is for. Why are the second and fourth parameters in our example both numSent? Can you explain what this means in terms of MPI_Scatter?
Run with varying numbers of processes from 1 through 8 by changing the -np flag. What issues do you see for certain numbers of processors?
What happens when you try more than 8 processors?
Be certain that you follow the results shown and can match them to the code example.

To Consider:

What previous data decomposition pattern is this similar to?

Warning

If you tried running this as suggested above, you can see that there are circumstances where scatter does not behave as you might intend. In other code that you write, you should add more checks to ensure that the size of your array and the number of processes will result in correct behavior.

15. Collective communication: Gather for message-passing data decomposition¶

If processes can independently work on portions of a larger data array using the geometric data decomposition pattern, the gather pattern can be used to ensure that each process sends a copy of its portion of the array back to the root, or conductor process. Thus, gather is the reverse of scatter. Here is the idea:

/* gather.c
  * ... illustrates the use of MPI_Gather()...
  */

#include <mpi.h>       // MPI
  #include <stdio.h>     // printf()
  #include <stdlib.h>    // malloc()

void print(int id, char* arrName, int* arr, int arrSize);

#define SIZE 3

int main(int argc, char** argv) {
    int  computeArray[SIZE];                          // array1
    int* gatherArray = NULL;                          // array2
    int  numProcs = -1, myRank = -1,
          totalGatheredVals = -1;

MPI_Init(&argc, &argv);                           // initialize
    MPI_Comm_size(MPI_COMM_WORLD, &numProcs);
    MPI_Comm_rank(MPI_COMM_WORLD, &myRank);
                                                      // all processes:
    for (int i = 0; i < SIZE; i++) {                  //  load array1 with
        computeArray[i] = myRank * 10 + i;             //   3 distinct values
    }

print(myRank, "computeArray", computeArray,       //  show array1
            SIZE);

if (myRank == 0) {                                // conductor:
        totalGatheredVals = SIZE * numProcs;           //  allocate array2
        gatherArray = (int*) malloc( totalGatheredVals * sizeof(int) );
    }

MPI_Gather(computeArray, SIZE, MPI_INT,           //  gather array1 vals
                gatherArray, SIZE, MPI_INT,           //   into array2
                0, MPI_COMM_WORLD);                   //   at conductor process

if (myRank == 0) {                                // conductor process:
        print(myRank, "gatherArray",                   //  show array2
              gatherArray, totalGatheredVals);
        free(gatherArray);                             // clean up
    }

MPI_Finalize();
    return 0;
  }

To do:

Find documentation for the MPI function MPI_Gather. Make sure that you know what each parameter is for. Why are the second and fourth parameters in our example both SIZE? Can you explain what this means in terms of MPI_Gather?
Run with varying numbers of processes by changing the -np flag.
Be certain that you follow the results shown and can match them to the code example.

You have attempted of activities on this page