src/smpi/smpi_coll.c

   1 /* $Id$tag */
   2
   3 /* smpi_coll.c -- various optimized routing for collectives                   */
   4
   5 /* Copyright (c) 2009 Stephane Genaud.                                        */
   6 /* All rights reserved.                                                       */
   7
   8 /* This program is free software; you can redistribute it and/or modify it
   9  *  * under the terms of the license (GNU LGPL) which comes with this package. */
  10
  11 #include <stdio.h>
  12 #include <string.h>
  13 #include <assert.h>
  14
  15 #include "private.h"
  16 #include "smpi_coll_private.h"
  17
  18 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_coll, smpi,
  19                                 "Logging specific to SMPI (coll)");
  20
  21 struct s_proc_tree {
  22   int PROCTREE_A;
  23   int numChildren;
  24   int * child;
  25   int parent;
  26   int me;
  27   int root;
  28   int isRoot;
  29 };
  30 typedef struct s_proc_tree * proc_tree_t;
  31
  32 /**
  33  * alloc and init
  34  **/
  35 static proc_tree_t alloc_tree(int arity) {
  36   proc_tree_t tree;
  37   int i;
  38
  39   tree = xbt_new(struct s_proc_tree, 1);
  40   tree->PROCTREE_A = arity;
  41   tree->isRoot = 0;
  42   tree->numChildren = 0;
  43   tree->child = xbt_new(int, arity);
  44   for(i = 0; i < arity; i++) {
  45     tree->child[i] = -1;
  46   }
  47   tree->root = -1;
  48   tree->parent = -1;
  49   return tree;
  50 }
  51
  52 /**
  53  * free
  54  **/
  55 static void free_tree(proc_tree_t tree) {
  56   xbt_free(tree->child );
  57   xbt_free(tree);
  58 }
  59
  60 /**
  61  * Build the tree depending on a process rank (index) and the group size (extent)
  62  * @param index the rank of the calling process
  63  * @param extent the total number of processes
  64  **/
  65 static void build_tree(int index, int extent, proc_tree_t* tree) {
  66   int places = (*tree)->PROCTREE_A * index;
  67   int i, ch, pr;
  68
  69   (*tree)->me = index;
  70   (*tree)->root = 0 ;
  71   for(i = 1; i <= (*tree)->PROCTREE_A; i++) {
  72     ++places;
  73     ch = (*tree)->PROCTREE_A * index + i + (*tree)->root;
  74     ch %= extent;
  75     if(places < extent) {
  76       (*tree)->child[i - 1] = ch;
  77       (*tree)->numChildren++;
  78     }
  79   }
  80   if(index == (*tree)->root) {
  81     (*tree)->isRoot = 1;
  82   } else {
  83     (*tree)->isRoot = 0;
  84     pr = (index - 1) / (*tree)->PROCTREE_A;
  85     (*tree)->parent = pr;
  86   }
  87 }
  88
  89 /**
  90  * bcast
  91  **/
  92 static void tree_bcast(void* buf, int count, MPI_Datatype datatype, int root, MPI_Comm comm, proc_tree_t tree) {
  93   int system_tag = 999;  // used negative int but smpi_create_request() declares this illegal (to be checked)
  94   int rank, i;
  95   MPI_Request* requests;
  96
  97   rank = smpi_comm_rank(comm);
  98   /* wait for data from my parent in the tree */
  99   if(!tree->isRoot) {
 100     DEBUG3("<%d> tree_bcast(): i am not root: recv from %d, tag=%d)",
 101            rank, tree->parent, system_tag + rank);
 102     smpi_mpi_recv(buf, count, datatype, tree->parent, system_tag + rank, comm, MPI_STATUS_IGNORE);
 103   }
 104   requests = xbt_new(MPI_Request, tree->numChildren);
 105   DEBUG2("<%d> creates %d requests (1 per child)", rank, tree->numChildren);
 106   /* iniates sends to ranks lower in the tree */
 107   for(i = 0; i < tree->numChildren; i++) {
 108     if(tree->child[i] == -1) {
 109       requests[i] = MPI_REQUEST_NULL;
 110     } else {
 111       DEBUG3("<%d> send to <%d>, tag=%d", rank, tree->child[i], system_tag + tree->child[i]);
 112       requests[i] = smpi_isend_init(buf, count, datatype, tree->child[i], system_tag + tree->child[i], comm);
 113     }
 114   }
 115   smpi_mpi_startall(tree->numChildren, requests);
 116   smpi_mpi_waitall(tree->numChildren, requests, MPI_STATUS_IGNORE);
 117   xbt_free(requests);
 118 }
 119
 120 /**
 121  * anti-bcast
 122  **/
 123 static void tree_antibcast(void* buf, int count, MPI_Datatype datatype, int root, MPI_Comm comm, proc_tree_t tree) {
 124   int system_tag = 999;  // used negative int but smpi_create_request() declares this illegal (to be checked)
 125   int rank, i;
 126   MPI_Request* requests;
 127
 128   rank = smpi_comm_rank(comm);
 129   // everyone sends to its parent, except root.
 130   if(!tree->isRoot) {
 131     DEBUG3("<%d> tree_antibcast(): i am not root: send to %d, tag=%d)",
 132            rank, tree->parent, system_tag + rank);
 133     smpi_mpi_send(buf, count, datatype, tree->parent, system_tag + rank, comm);
 134   }
 135   //every one receives as many messages as it has children
 136   requests = xbt_new(MPI_Request, tree->numChildren);
 137   DEBUG2("<%d> creates %d requests (1 per child)", rank, tree->numChildren);
 138   for(i = 0; i < tree->numChildren; i++) {
 139     if(tree->child[i] == -1) {
 140       requests[i] = MPI_REQUEST_NULL;
 141     } else {
 142       DEBUG3("<%d> recv from <%d>, tag=%d", rank, tree->child[i], system_tag + tree->child[i]);
 143       requests[i] = smpi_irecv_init(buf, count, datatype, tree->child[i], system_tag + tree->child[i], comm);
 144     }
 145   }
 146   smpi_mpi_startall(tree->numChildren, requests);
 147   smpi_mpi_waitall(tree->numChildren, requests, MPI_STATUS_IGNORE);
 148   xbt_free(requests);
 149 }
 150
 151 /**
 152  * bcast with a binary, ternary, or whatever tree ..
 153  **/
 154 void nary_tree_bcast(void* buf, int count, MPI_Datatype datatype, int root, MPI_Comm comm, int arity) {
 155   proc_tree_t tree = alloc_tree(arity);
 156   int rank, size;
 157
 158   rank = smpi_comm_rank(comm);
 159   size = smpi_comm_size(comm);
 160   build_tree(rank, size, &tree);
 161   tree_bcast(buf, count, datatype, root, comm, tree);
 162   free_tree(tree);
 163 }
 164
 165 /**
 166  * barrier with a binary, ternary, or whatever tree ..
 167  **/
 168 void nary_tree_barrier(MPI_Comm comm, int arity) {
 169   proc_tree_t tree = alloc_tree( arity );
 170   int rank, size;
 171   char dummy='$';
 172
 173   rank = smpi_comm_rank(comm);
 174   size = smpi_comm_size(comm);
 175   build_tree(rank, size, &tree);
 176   tree_antibcast(&dummy, 1, MPI_CHAR, 0, comm, tree);
 177   tree_bcast(&dummy, 1, MPI_CHAR, 0, comm, tree);
 178   free_tree(tree);
 179 }
 180
 181 /**
 182  * Alltoall Bruck
 183  *
 184  * Openmpi calls this routine when the message size sent to each rank < 2000 bytes and size < 12
 185  **/
 186 int smpi_coll_tuned_alltoall_bruck(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, MPI_Comm comm) {
 187   int system_tag = 777;
 188   int i, rank, size, err, count;
 189   MPI_Aint lb, sendextent, recvextent;
 190   MPI_Request* requests;
 191
 192   // FIXME: check implementation
 193   rank = smpi_comm_rank(comm);
 194   size = smpi_comm_size(comm);
 195   DEBUG1("<%d> algorithm alltoall_bruck() called.", rank);
 196   err = smpi_datatype_extent(sendtype, &lb, &sendextent);
 197   err = smpi_datatype_extent(recvtype, &lb, &recvextent);
 198   /* Local copy from self */
 199   err = smpi_datatype_copy(&((char*)sendbuf)[rank * sendextent], sendcount, sendtype, &((char*)recvbuf)[rank * recvextent], recvcount, recvtype);
 200   if(err == MPI_SUCCESS && size > 1) {
 201     /* Initiate all send/recv to/from others. */
 202     requests = xbt_new(MPI_Request, 2 * (size - 1));
 203     count = 0;
 204     /* Create all receives that will be posted first */
 205     for(i = 0; i < size; ++i) {
 206       if(i == rank) {
 207         DEBUG3("<%d> skip request creation [src = %d, recvcount = %d]", rank, i, recvcount);
 208         continue;
 209       }
 210       requests[count] = smpi_irecv_init(&((char*)recvbuf)[i * recvextent], recvcount, recvtype, i, system_tag, comm);
 211       count++;
 212     }
 213     /* Now create all sends  */
 214     for(i = 0; i < size; ++i) {
 215       if(i == rank) {
 216         DEBUG3("<%d> skip request creation [dst = %d, sendcount = %d]", rank, i, sendcount);
 217         continue;
 218       }
 219       requests[count] = smpi_isend_init(&((char*)sendbuf)[i * sendextent], sendcount, sendtype, i, system_tag, comm);
 220       count++;
 221     }
 222     /* Wait for them all.*/
 223     smpi_mpi_startall(count, requests);
 224     DEBUG2("<%d> wait for %d requests", rank, count);
 225     smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE);
 226     xbt_free(requests);
 227   }
 228   return MPI_SUCCESS;
 229 }
 230
 231 /**
 232  * Alltoall basic_linear
 233  **/
 234 int smpi_coll_tuned_alltoall_basic_linear(void *sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, MPI_Comm comm) {
 235   int system_tag = 888;
 236   int i, rank, size, err, count;
 237   MPI_Aint lb, sendinc, recvinc;
 238   MPI_Request *requests;
 239
 240   /* Initialize. */
 241   rank = smpi_comm_rank(comm);
 242   size = smpi_comm_size(comm);
 243   DEBUG1("<%d> algorithm alltoall_basic_linear() called.", rank);
 244   err = smpi_datatype_extent(sendtype, &lb, &sendinc);
 245   err = smpi_datatype_extent(recvtype, &lb, &recvinc);
 246   sendinc *= sendcount;
 247   recvinc *= recvcount;
 248   /* simple optimization */
 249   err = smpi_datatype_copy(&((char*)sendbuf)[rank * sendinc], sendcount, sendtype, &((char*)recvbuf)[rank * recvinc], recvcount, recvtype);
 250   if(err == MPI_SUCCESS && size > 1) {
 251     /* Initiate all send/recv to/from others. */
 252     requests = xbt_new(MPI_Request, 2 * (size - 1));
 253     /* Post all receives first -- a simple optimization */
 254     count = 0;
 255     for(i = (rank + 1) % size; i != rank; i = (i + 1) % size) {
 256       requests[count] = smpi_irecv_init(&((char*)recvbuf)[i * recvinc], recvcount, recvtype, i, system_tag, comm);
 257       count++;
 258     }
 259     /* Now post all sends in reverse order
 260      *   - We would like to minimize the search time through message queue
 261      *     when messages actually arrive in the order in which they were posted.
 262      * TODO: check the previous assertion
 263      */
 264     for(i = (rank + size - 1) % size; i != rank; i = (i + size - 1) % size ) {
 265       requests[count] = smpi_isend_init(&((char*)sendbuf)[i * sendinc], sendcount, sendtype, i, system_tag, comm);
 266       count++;
 267     }
 268     /* Wait for them all.*/
 269     smpi_mpi_startall(count, requests);
 270     DEBUG2("<%d> wait for %d requests", rank, count);
 271     smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE);
 272     xbt_free(requests);
 273   }
 274   return err;
 275 }
 276
 277 /**
 278  * Alltoall pairwise
 279  *
 280  * this algorithm performs size steps (1<=s<=size) and
 281  * at each step s, a process p sends iand receive to.from a unique distinct remote process
 282  * size=5 : s=1:  4->0->1, 0->1->2, 1->2->3, ...
 283  *          s=2:  3->0->2, 4->1->3, 0->2->4, 1->3->0 , 2->4->1
 284  *          ....
 285  * Openmpi calls this routine when the message size sent to each rank is greater than 3000 bytes
 286  **/
 287 int smpi_coll_tuned_alltoall_pairwise(void* sendbuf, int sendcount, MPI_Datatype sendtype, void* recvbuf, int recvcount, MPI_Datatype recvtype, MPI_Comm comm) {
 288   int system_tag = 999;
 289   int rank, size, step, sendto, recvfrom, sendsize, recvsize;
 290
 291   rank = smpi_comm_rank(comm);
 292   size = smpi_comm_size(comm);
 293   DEBUG1("<%d> algorithm alltoall_pairwise() called.", rank);
 294   sendsize = smpi_datatype_size(sendtype);
 295   recvsize = smpi_datatype_size(recvtype);
 296   /* Perform pairwise exchange - starting from 1 so the local copy is last */
 297   for(step = 1; step < size + 1; step++) {
 298     /* who do we talk to in this step? */
 299     sendto  = (rank+step)%size;
 300     recvfrom = (rank+size-step)%size;
 301     /* send and receive */
 302     smpi_mpi_sendrecv(&((char*)sendbuf)[sendto * sendsize * sendcount], sendcount, sendtype, sendto, system_tag, &((char*)recvbuf)[recvfrom * recvsize * recvcount], recvcount, recvtype, recvfrom, system_tag, comm, MPI_STATUS_IGNORE);
 303   }
 304   return MPI_SUCCESS;
 305 }
 306
 307 int smpi_coll_basic_alltoallv(void* sendbuf, int* sendcounts, int* senddisps, MPI_Datatype sendtype, void* recvbuf, int *recvcounts, int* recvdisps, MPI_Datatype recvtype, MPI_Comm comm) {
 308   int system_tag = 889;
 309   int i, rank, size, err, count;
 310   MPI_Aint lb, sendextent, recvextent;
 311   MPI_Request* requests;
 312
 313   /* Initialize. */
 314   rank = smpi_comm_rank(comm);
 315   size = smpi_comm_size(comm);
 316   DEBUG1("<%d> algorithm basic_alltoallv() called.", rank);
 317   err = smpi_datatype_extent(sendtype, &lb, &sendextent);
 318   err = smpi_datatype_extent(recvtype, &lb, &recvextent);
 319   /* Local copy from self */
 320   err = smpi_datatype_copy(&((char*)sendbuf)[senddisps[rank] * sendextent], sendcounts[rank], sendtype, &((char*)recvbuf)[recvdisps[rank] * recvextent], recvcounts[rank], recvtype);
 321   if(err == MPI_SUCCESS && size > 1) {
 322     /* Initiate all send/recv to/from others. */
 323     requests = xbt_new(MPI_Request, 2 * (size - 1));
 324     count = 0;
 325     /* Create all receives that will be posted first */
 326     for(i = 0; i < size; ++i) {
 327       if(i == rank || recvcounts[i] == 0) {
 328         DEBUG3("<%d> skip request creation [src = %d, recvcounts[src] = %d]", rank, i, recvcounts[i]);
 329         continue;
 330       }
 331       requests[count] = smpi_irecv_init(&((char*)recvbuf)[recvdisps[i] * recvextent], recvcounts[i], recvtype, i, system_tag, comm);
 332       count++;
 333     }
 334     /* Now create all sends  */
 335     for(i = 0; i < size; ++i) {
 336       if(i == rank || sendcounts[i] == 0) {
 337         DEBUG3("<%d> skip request creation [dst = %d, sendcounts[dst] = %d]", rank, i, sendcounts[i]);
 338         continue;
 339       }
 340       requests[count] = smpi_isend_init(&((char*)sendbuf)[senddisps[i] * sendextent], sendcounts[i], sendtype, i, system_tag, comm);
 341       count++;
 342     }
 343     /* Wait for them all.*/
 344     smpi_mpi_startall(count, requests);
 345     DEBUG2("<%d> wait for %d requests", rank, count);
 346     smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE);
 347     xbt_free(requests);
 348   }
 349   return err;
 350 }