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