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