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 "colls/colls.h"
  15
  16 s_mpi_coll_description_t mpi_coll_allgather_description[] = {
  17   {"default",
  18    "allgather default collective",
  19    smpi_mpi_allgather},
  20 COLL_ALLGATHERS(COLL_DESCRIPTION, COLL_COMMA),
  21   {NULL, NULL, NULL}      /* this array must be NULL terminated */
  22 };
  23
  24 s_mpi_coll_description_t mpi_coll_allreduce_description[] = {
  25   {"default",
  26    "allreduce default collective",
  27    smpi_mpi_allreduce},
  28 COLL_ALLREDUCES(COLL_DESCRIPTION, COLL_COMMA),
  29   {NULL, NULL, NULL}      /* this array must be NULL terminated */
  30 };
  31
  32 s_mpi_coll_description_t mpi_coll_alltoall_description[] = {
  33   {"ompi",
  34    "Ompi alltoall default collective",
  35    smpi_coll_tuned_alltoall_ompi},
  36 COLL_ALLTOALLS(COLL_DESCRIPTION, COLL_COMMA),
  37   {"bruck",
  38    "Alltoall Bruck (SG) collective",
  39    smpi_coll_tuned_alltoall_bruck},
  40   {"basic_linear",
  41    "Alltoall basic linear (SG) collective",
  42    smpi_coll_tuned_alltoall_basic_linear},
  43   {NULL, NULL, NULL}      /* this array must be NULL terminated */
  44 };
  45
  46 s_mpi_coll_description_t mpi_coll_alltoallv_description[] = {
  47   {"default",
  48    "Ompi alltoallv default collective",
  49    smpi_coll_basic_alltoallv},
  50 COLL_ALLTOALLVS(COLL_DESCRIPTION, COLL_COMMA),
  51   {NULL, NULL, NULL}      /* this array must be NULL terminated */
  52 };
  53
  54 s_mpi_coll_description_t mpi_coll_bcast_description[] = {
  55   {"default",
  56    "allgather default collective",
  57    smpi_mpi_bcast},
  58 COLL_BCASTS(COLL_DESCRIPTION, COLL_COMMA),
  59   {NULL, NULL, NULL}      /* this array must be NULL terminated */
  60 };
  61
  62 s_mpi_coll_description_t mpi_coll_reduce_description[] = {
  63   {"default",
  64    "allgather default collective",
  65    smpi_mpi_reduce},
  66 COLL_REDUCES(COLL_DESCRIPTION, COLL_COMMA),
  67   {NULL, NULL, NULL}      /* this array must be NULL terminated */
  68 };
  69
  70
  71
  72 /** Displays the long description of all registered models, and quit */
  73 void coll_help(const char *category, s_mpi_coll_description_t * table)
  74 {
  75   int i;
  76   printf("Long description of the %s models accepted by this simulator:\n",
  77          category);
  78   for (i = 0; table[i].name; i++)
  79     printf("  %s: %s\n", table[i].name, table[i].description);
  80 }
  81
  82 int find_coll_description(s_mpi_coll_description_t * table,
  83                            const char *name)
  84 {
  85   int i;
  86   char *name_list = NULL;
  87
  88   for (i = 0; table[i].name; i++)
  89     if (!strcmp(name, table[i].name)) {
  90       return i;
  91     }
  92   name_list = strdup(table[0].name);
  93   for (i = 1; table[i].name; i++) {
  94     name_list =
  95         xbt_realloc(name_list,
  96                     strlen(name_list) + strlen(table[i].name) + 3);
  97     strcat(name_list, ", ");
  98     strcat(name_list, table[i].name);
  99   }
 100   xbt_die("Model '%s' is invalid! Valid models are: %s.", name, name_list);
 101   return -1;
 102 }
 103
 104
 105
 106 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_coll, smpi,
 107                                 "Logging specific to SMPI (coll)");
 108
 109 int (*mpi_coll_allgather_fun)(void *, int, MPI_Datatype, void*, int, MPI_Datatype, MPI_Comm);
 110 int (*mpi_coll_allreduce_fun)(void *sbuf, void *rbuf, int rcount, MPI_Datatype dtype, MPI_Op op, MPI_Comm comm);
 111 int (*mpi_coll_alltoall_fun)(void *, int, MPI_Datatype, void*, int, MPI_Datatype, MPI_Comm);
 112 int (*mpi_coll_alltoallv_fun)(void *, int*, int*, MPI_Datatype, void*, int*, int*, MPI_Datatype, MPI_Comm);
 113 int (*mpi_coll_bcast_fun)(void *buf, int count, MPI_Datatype datatype, int root, MPI_Comm com);
 114 int (*mpi_coll_reduce_fun)(void *buf, void *rbuf, int count, MPI_Datatype datatype, MPI_Op op, int root, MPI_Comm comm);
 115
 116 struct s_proc_tree {
 117   int PROCTREE_A;
 118   int numChildren;
 119   int *child;
 120   int parent;
 121   int me;
 122   int root;
 123   int isRoot;
 124 };
 125 typedef struct s_proc_tree *proc_tree_t;
 126
 127 /**
 128  * alloc and init
 129  **/
 130 static proc_tree_t alloc_tree(int arity)
 131 {
 132   proc_tree_t tree;
 133   int i;
 134
 135   tree = xbt_new(struct s_proc_tree, 1);
 136   tree->PROCTREE_A = arity;
 137   tree->isRoot = 0;
 138   tree->numChildren = 0;
 139   tree->child = xbt_new(int, arity);
 140   for (i = 0; i < arity; i++) {
 141     tree->child[i] = -1;
 142   }
 143   tree->root = -1;
 144   tree->parent = -1;
 145   return tree;
 146 }
 147
 148 /**
 149  * free
 150  **/
 151 static void free_tree(proc_tree_t tree)
 152 {
 153   xbt_free(tree->child);
 154   xbt_free(tree);
 155 }
 156
 157 /**
 158  * Build the tree depending on a process rank (index) and the group size (extent)
 159  * @param root the rank of the tree root
 160  * @param rank the rank of the calling process
 161  * @param size the total number of processes
 162  **/
 163 static void build_tree(int root, int rank, int size, proc_tree_t * tree)
 164 {
 165   int index = (rank - root + size) % size;
 166   int firstChildIdx = index * (*tree)->PROCTREE_A + 1;
 167   int i;
 168
 169   (*tree)->me = rank;
 170   (*tree)->root = root;
 171
 172   for (i = 0; i < (*tree)->PROCTREE_A && firstChildIdx + i < size; i++) {
 173     (*tree)->child[i] = (firstChildIdx + i + root) % size;
 174     (*tree)->numChildren++;
 175   }
 176   if (rank == root) {
 177     (*tree)->isRoot = 1;
 178   } else {
 179     (*tree)->isRoot = 0;
 180     (*tree)->parent = (((index - 1) / (*tree)->PROCTREE_A) + root) % size;
 181   }
 182 }
 183
 184 /**
 185  * bcast
 186  **/
 187 static void tree_bcast(void *buf, int count, MPI_Datatype datatype,
 188                        MPI_Comm comm, proc_tree_t tree)
 189 {
 190   int system_tag = 999;         // used negative int but smpi_create_request() declares this illegal (to be checked)
 191   int rank, i;
 192   MPI_Request *requests;
 193
 194   rank = smpi_comm_rank(comm);
 195   /* wait for data from my parent in the tree */
 196   if (!tree->isRoot) {
 197     XBT_DEBUG("<%d> tree_bcast(): i am not root: recv from %d, tag=%d)",
 198            rank, tree->parent, system_tag + rank);
 199     smpi_mpi_recv(buf, count, datatype, tree->parent, system_tag + rank,
 200                   comm, MPI_STATUS_IGNORE);
 201   }
 202   requests = xbt_new(MPI_Request, tree->numChildren);
 203   XBT_DEBUG("<%d> creates %d requests (1 per child)", rank,
 204          tree->numChildren);
 205   /* iniates sends to ranks lower in the tree */
 206   for (i = 0; i < tree->numChildren; i++) {
 207     if (tree->child[i] == -1) {
 208       requests[i] = MPI_REQUEST_NULL;
 209     } else {
 210       XBT_DEBUG("<%d> send to <%d>, tag=%d", rank, tree->child[i],
 211              system_tag + tree->child[i]);
 212       requests[i] =
 213           smpi_isend_init(buf, count, datatype, tree->child[i],
 214                           system_tag + tree->child[i], comm);
 215     }
 216   }
 217   smpi_mpi_startall(tree->numChildren, requests);
 218   smpi_mpi_waitall(tree->numChildren, requests, MPI_STATUS_IGNORE);
 219   xbt_free(requests);
 220 }
 221
 222 /**
 223  * anti-bcast
 224  **/
 225 static void tree_antibcast(void *buf, int count, MPI_Datatype datatype,
 226                            MPI_Comm comm, proc_tree_t tree)
 227 {
 228   int system_tag = 999;         // used negative int but smpi_create_request() declares this illegal (to be checked)
 229   int rank, i;
 230   MPI_Request *requests;
 231
 232   rank = smpi_comm_rank(comm);
 233   // everyone sends to its parent, except root.
 234   if (!tree->isRoot) {
 235     XBT_DEBUG("<%d> tree_antibcast(): i am not root: send to %d, tag=%d)",
 236            rank, tree->parent, system_tag + rank);
 237     smpi_mpi_send(buf, count, datatype, tree->parent, system_tag + rank,
 238                   comm);
 239   }
 240   //every one receives as many messages as it has children
 241   requests = xbt_new(MPI_Request, tree->numChildren);
 242   XBT_DEBUG("<%d> creates %d requests (1 per child)", rank,
 243          tree->numChildren);
 244   for (i = 0; i < tree->numChildren; i++) {
 245     if (tree->child[i] == -1) {
 246       requests[i] = MPI_REQUEST_NULL;
 247     } else {
 248       XBT_DEBUG("<%d> recv from <%d>, tag=%d", rank, tree->child[i],
 249              system_tag + tree->child[i]);
 250       requests[i] =
 251           smpi_irecv_init(buf, count, datatype, tree->child[i],
 252                           system_tag + tree->child[i], comm);
 253     }
 254   }
 255   smpi_mpi_startall(tree->numChildren, requests);
 256   smpi_mpi_waitall(tree->numChildren, requests, MPI_STATUS_IGNORE);
 257   xbt_free(requests);
 258 }
 259
 260 /**
 261  * bcast with a binary, ternary, or whatever tree ..
 262  **/
 263 void nary_tree_bcast(void *buf, int count, MPI_Datatype datatype, int root,
 264                      MPI_Comm comm, int arity)
 265 {
 266   proc_tree_t tree = alloc_tree(arity);
 267   int rank, size;
 268
 269   rank = smpi_comm_rank(comm);
 270   size = smpi_comm_size(comm);
 271   build_tree(root, rank, size, &tree);
 272   tree_bcast(buf, count, datatype, comm, tree);
 273   free_tree(tree);
 274 }
 275
 276 /**
 277  * barrier with a binary, ternary, or whatever tree ..
 278  **/
 279 void nary_tree_barrier(MPI_Comm comm, int arity)
 280 {
 281   proc_tree_t tree = alloc_tree(arity);
 282   int rank, size;
 283   char dummy = '$';
 284
 285   rank = smpi_comm_rank(comm);
 286   size = smpi_comm_size(comm);
 287   build_tree(0, rank, size, &tree);
 288   tree_antibcast(&dummy, 1, MPI_CHAR, comm, tree);
 289   tree_bcast(&dummy, 1, MPI_CHAR, comm, tree);
 290   free_tree(tree);
 291 }
 292
 293 int smpi_coll_tuned_alltoall_ompi(void *sendbuf, int sendcount,
 294                                    MPI_Datatype sendtype, void *recvbuf,
 295                                    int recvcount, MPI_Datatype recvtype,
 296                                    MPI_Comm comm)
 297 {
 298   int size, sendsize;
 299   size = smpi_comm_size(comm);
 300   sendsize = smpi_datatype_size(sendtype) * sendcount;
 301   if (sendsize < 200 && size > 12) {
 302     return
 303         smpi_coll_tuned_alltoall_bruck(sendbuf, sendcount, sendtype,
 304                                        recvbuf, recvcount, recvtype,
 305                                        comm);
 306   } else if (sendsize < 3000) {
 307     return
 308         smpi_coll_tuned_alltoall_basic_linear(sendbuf, sendcount,
 309                                               sendtype, recvbuf,
 310                                               recvcount, recvtype, comm);
 311   } else {
 312     return
 313         smpi_coll_tuned_alltoall_ring(sendbuf, sendcount, sendtype,
 314                                       recvbuf, recvcount, recvtype,
 315                                       comm);
 316   }
 317 }
 318
 319 /**
 320  * Alltoall Bruck
 321  *
 322  * Openmpi calls this routine when the message size sent to each rank < 2000 bytes and size < 12
 323  * FIXME: uh, check smpi_pmpi again, but this routine is called for > 12, not
 324  * less...
 325  **/
 326 int smpi_coll_tuned_alltoall_bruck(void *sendbuf, int sendcount,
 327                                    MPI_Datatype sendtype, void *recvbuf,
 328                                    int recvcount, MPI_Datatype recvtype,
 329                                    MPI_Comm comm)
 330 {
 331   int system_tag = 777;
 332   int i, rank, size, err, count;
 333   MPI_Aint lb;
 334   MPI_Aint sendext = 0;
 335   MPI_Aint recvext = 0;
 336   MPI_Request *requests;
 337
 338   // FIXME: check implementation
 339   rank = smpi_comm_rank(comm);
 340   size = smpi_comm_size(comm);
 341   XBT_DEBUG("<%d> algorithm alltoall_bruck() called.", rank);
 342   err = smpi_datatype_extent(sendtype, &lb, &sendext);
 343   err = smpi_datatype_extent(recvtype, &lb, &recvext);
 344   /* Local copy from self */
 345   err =
 346       smpi_datatype_copy((char *)sendbuf + rank * sendcount * sendext,
 347                          sendcount, sendtype,
 348                          (char *)recvbuf + rank * recvcount * recvext,
 349                          recvcount, recvtype);
 350   if (err == MPI_SUCCESS && size > 1) {
 351     /* Initiate all send/recv to/from others. */
 352     requests = xbt_new(MPI_Request, 2 * (size - 1));
 353     count = 0;
 354     /* Create all receives that will be posted first */
 355     for (i = 0; i < size; ++i) {
 356       if (i == rank) {
 357         XBT_DEBUG("<%d> skip request creation [src = %d, recvcount = %d]",
 358                rank, i, recvcount);
 359         continue;
 360       }
 361       requests[count] =
 362           smpi_irecv_init((char *)recvbuf + i * recvcount * recvext, recvcount,
 363                           recvtype, i, system_tag, comm);
 364       count++;
 365     }
 366     /* Now create all sends  */
 367     for (i = 0; i < size; ++i) {
 368       if (i == rank) {
 369         XBT_DEBUG("<%d> skip request creation [dst = %d, sendcount = %d]",
 370                rank, i, sendcount);
 371         continue;
 372       }
 373       requests[count] =
 374           smpi_isend_init((char *)sendbuf + i * sendcount * sendext, sendcount,
 375                           sendtype, i, system_tag, comm);
 376       count++;
 377     }
 378     /* Wait for them all. */
 379     smpi_mpi_startall(count, requests);
 380     XBT_DEBUG("<%d> wait for %d requests", rank, count);
 381     smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE);
 382     xbt_free(requests);
 383   }
 384   return MPI_SUCCESS;
 385 }
 386
 387 /**
 388  * Alltoall basic_linear (STARMPI:alltoall-simple)
 389  **/
 390 int smpi_coll_tuned_alltoall_basic_linear(void *sendbuf, int sendcount,
 391                                           MPI_Datatype sendtype,
 392                                           void *recvbuf, int recvcount,
 393                                           MPI_Datatype recvtype,
 394                                           MPI_Comm comm)
 395 {
 396   int system_tag = 888;
 397   int i, rank, size, err, count;
 398   MPI_Aint lb = 0, sendext = 0, recvext = 0;
 399   MPI_Request *requests;
 400
 401   /* Initialize. */
 402   rank = smpi_comm_rank(comm);
 403   size = smpi_comm_size(comm);
 404   XBT_DEBUG("<%d> algorithm alltoall_basic_linear() called.", rank);
 405   err = smpi_datatype_extent(sendtype, &lb, &sendext);
 406   err = smpi_datatype_extent(recvtype, &lb, &recvext);
 407   /* simple optimization */
 408   err = smpi_datatype_copy((char *)sendbuf + rank * sendcount * sendext,
 409                            sendcount, sendtype,
 410                            (char *)recvbuf + rank * recvcount * recvext,
 411                            recvcount, recvtype);
 412   if (err == MPI_SUCCESS && size > 1) {
 413     /* Initiate all send/recv to/from others. */
 414     requests = xbt_new(MPI_Request, 2 * (size - 1));
 415     /* Post all receives first -- a simple optimization */
 416     count = 0;
 417     for (i = (rank + 1) % size; i != rank; i = (i + 1) % size) {
 418       requests[count] =
 419           smpi_irecv_init((char *)recvbuf + i * recvcount * recvext, recvcount,
 420                           recvtype, i, system_tag, comm);
 421       count++;
 422     }
 423     /* Now post all sends in reverse order
 424      *   - We would like to minimize the search time through message queue
 425      *     when messages actually arrive in the order in which they were posted.
 426      * TODO: check the previous assertion
 427      */
 428     for (i = (rank + size - 1) % size; i != rank; i = (i + size - 1) % size) {
 429       requests[count] =
 430           smpi_isend_init((char *)sendbuf + i * sendcount * sendext, sendcount,
 431                           sendtype, i, system_tag, comm);
 432       count++;
 433     }
 434     /* Wait for them all. */
 435     smpi_mpi_startall(count, requests);
 436     XBT_DEBUG("<%d> wait for %d requests", rank, count);
 437     smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE);
 438     xbt_free(requests);
 439   }
 440   return err;
 441 }
 442
 443 int smpi_coll_basic_alltoallv(void *sendbuf, int *sendcounts,
 444                               int *senddisps, MPI_Datatype sendtype,
 445                               void *recvbuf, int *recvcounts,
 446                               int *recvdisps, MPI_Datatype recvtype,
 447                               MPI_Comm comm)
 448 {
 449   int system_tag = 889;
 450   int i, rank, size, err, count;
 451   MPI_Aint lb = 0, sendext = 0, recvext = 0;
 452   MPI_Request *requests;
 453
 454   /* Initialize. */
 455   rank = smpi_comm_rank(comm);
 456   size = smpi_comm_size(comm);
 457   XBT_DEBUG("<%d> algorithm basic_alltoallv() called.", rank);
 458   err = smpi_datatype_extent(sendtype, &lb, &sendext);
 459   err = smpi_datatype_extent(recvtype, &lb, &recvext);
 460   /* Local copy from self */
 461   err =
 462       smpi_datatype_copy((char *)sendbuf + senddisps[rank] * sendext,
 463                          sendcounts[rank], sendtype,
 464                          (char *)recvbuf + recvdisps[rank] * recvext,
 465                          recvcounts[rank], recvtype);
 466   if (err == MPI_SUCCESS && size > 1) {
 467     /* Initiate all send/recv to/from others. */
 468     requests = xbt_new(MPI_Request, 2 * (size - 1));
 469     count = 0;
 470     /* Create all receives that will be posted first */
 471     for (i = 0; i < size; ++i) {
 472       if (i == rank || recvcounts[i] == 0) {
 473         XBT_DEBUG
 474             ("<%d> skip request creation [src = %d, recvcounts[src] = %d]",
 475              rank, i, recvcounts[i]);
 476         continue;
 477       }
 478       requests[count] =
 479           smpi_irecv_init((char *)recvbuf + recvdisps[i] * recvext,
 480                           recvcounts[i], recvtype, i, system_tag, comm);
 481       count++;
 482     }
 483     /* Now create all sends  */
 484     for (i = 0; i < size; ++i) {
 485       if (i == rank || sendcounts[i] == 0) {
 486         XBT_DEBUG
 487             ("<%d> skip request creation [dst = %d, sendcounts[dst] = %d]",
 488              rank, i, sendcounts[i]);
 489         continue;
 490       }
 491       requests[count] =
 492           smpi_isend_init((char *)sendbuf + senddisps[i] * sendext,
 493                           sendcounts[i], sendtype, i, system_tag, comm);
 494       count++;
 495     }
 496     /* Wait for them all. */
 497     smpi_mpi_startall(count, requests);
 498     XBT_DEBUG("<%d> wait for %d requests", rank, count);
 499     smpi_mpi_waitall(count, requests, MPI_STATUS_IGNORE);
 500     xbt_free(requests);
 501   }
 502   return err;
 503 }