src/smpi/smpi_coll.c

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