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