src/smpi/smpi_global.c

   1 /* Copyright (c) 2007-2014. The SimGrid Team.
   2  * All rights reserved.                                                     */
   3
   4 /* This program is free software; you can redistribute it and/or modify it
   5  * under the terms of the license (GNU LGPL) which comes with this package. */
   6
   7 #include "private.h"
   8 #include "smpi_mpi_dt_private.h"
   9 #include "mc/mc.h"
  10 #include "xbt/replay.h"
  11 #include "surf/surf.h"
  12 #include "simix/smx_private.h"
  13 #include "simgrid/sg_config.h"
  14
  15 #include <float.h>              /* DBL_MAX */
  16 #include <stdint.h>
  17 #include <stdio.h>
  18 #include <stdlib.h>
  19
  20 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_kernel, smpi,
  21                                 "Logging specific to SMPI (kernel)");
  22
  23 typedef struct s_smpi_process_data {
  24   double simulated;
  25   int *argc;
  26   char ***argv;
  27   smx_rdv_t mailbox;
  28   smx_rdv_t mailbox_small;
  29   xbt_os_timer_t timer;
  30   MPI_Comm comm_self;
  31   void *data;                   /* user data */
  32   int index;
  33   char state;
  34   int sampling;                 /* inside an SMPI_SAMPLE_ block? */
  35 } s_smpi_process_data_t;
  36
  37 static smpi_process_data_t *process_data = NULL;
  38 static int process_count = 0;
  39
  40 MPI_Comm MPI_COMM_WORLD = MPI_COMM_NULL;
  41 int MPI_UNIVERSE_SIZE;
  42
  43 MPI_Errhandler *MPI_ERRORS_RETURN = NULL;
  44 MPI_Errhandler *MPI_ERRORS_ARE_FATAL = NULL;
  45 MPI_Errhandler *MPI_ERRHANDLER_NULL = NULL;
  46
  47 #define MAILBOX_NAME_MAXLEN (5 + sizeof(int) * 2 + 1)
  48
  49 static char *get_mailbox_name(char *str, int index)
  50 {
  51   snprintf(str, MAILBOX_NAME_MAXLEN, "SMPI-%0*x", (int) (sizeof(int) * 2),
  52            index);
  53   return str;
  54 }
  55
  56 static char *get_mailbox_name_small(char *str, int index)
  57 {
  58   snprintf(str, MAILBOX_NAME_MAXLEN, "small%0*x", (int) (sizeof(int) * 2),
  59            index);
  60   return str;
  61 }
  62
  63 void smpi_process_init(int *argc, char ***argv)
  64 {
  65   int index;
  66   smpi_process_data_t data;
  67   smx_process_t proc;
  68
  69   if (argc && argv) {
  70     proc = SIMIX_process_self();
  71     index = atoi((*argv)[1]);
  72 #ifdef SMPI_F2C
  73     smpi_current_rank = index;
  74 #endif
  75     data = smpi_process_remote_data(index);
  76     simcall_process_set_data(proc, data);
  77     if (*argc > 2) {
  78       free((*argv)[1]);
  79       memmove(&(*argv)[1], &(*argv)[2], sizeof(char *) * (*argc - 2));
  80       (*argv)[(*argc) - 1] = NULL;
  81     }
  82     (*argc)--;
  83     data->argc = argc;
  84     data->argv = argv;
  85     // set the process attached to the mailbox
  86     simcall_rdv_set_receiver(data->mailbox_small, proc);
  87     XBT_DEBUG("<%d> New process in the game: %p", index, proc);
  88   }
  89 }
  90
  91 void smpi_process_destroy(void)
  92 {
  93   int index = smpi_process_index();
  94   process_data[index]->state = SMPI_FINALIZED;
  95   XBT_DEBUG("<%d> Process left the game", index);
  96 }
  97
  98 /**
  99  * @brief Prepares the current process for termination.
 100  */
 101 void smpi_process_finalize(void)
 102 {
 103   int i;
 104   int size = smpi_comm_size(MPI_COMM_WORLD);
 105   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 106   /* All non-root send & receive zero-length message. */
 107   if (rank > 0) {
 108     smpi_mpi_ssend (NULL, 0, MPI_BYTE, 0,
 109                     COLL_TAG_BARRIER,
 110                     MPI_COMM_WORLD);
 111     smpi_mpi_recv (NULL, 0, MPI_BYTE, 0,
 112                     COLL_TAG_BARRIER,
 113                     MPI_COMM_WORLD, MPI_STATUS_IGNORE);
 114   }
 115   /* The root collects and broadcasts the messages. */
 116   else {
 117     MPI_Request* requests;
 118     requests = (MPI_Request*)malloc( size * sizeof(MPI_Request) );
 119     for (i = 1; i < size; ++i) {
 120       requests[i] = smpi_mpi_irecv(NULL, 0, MPI_BYTE, MPI_ANY_SOURCE,
 121                                    COLL_TAG_BARRIER, MPI_COMM_WORLD
 122                                    );
 123     }
 124     smpi_mpi_waitall( size-1, requests+1, MPI_STATUSES_IGNORE );
 125     for (i = 1; i < size; ++i) {
 126       requests[i] = smpi_mpi_issend(NULL, 0, MPI_BYTE, i,
 127                                    COLL_TAG_BARRIER,
 128                                    MPI_COMM_WORLD
 129                                    );
 130     }
 131     smpi_mpi_waitall( size-1, requests+1, MPI_STATUSES_IGNORE );
 132     free( requests );
 133   }
 134
 135 }
 136
 137 /**
 138  * @brief Check if a process is finalized
 139  */
 140 int smpi_process_finalized()
 141 {
 142   int index = smpi_process_index();
 143     if (index != MPI_UNDEFINED)
 144       return (process_data[index]->state == SMPI_FINALIZED);
 145     else
 146       return 0;
 147 }
 148
 149 /**
 150  * @brief Check if a process is initialized
 151  */
 152 int smpi_process_initialized(void)
 153 {
 154   int index = smpi_process_index();
 155   return ( (index != MPI_UNDEFINED)
 156           && (process_data[index]->state == SMPI_INITIALIZED));
 157 }
 158
 159 /**
 160  * @brief Mark a process as initialized (=MPI_Init called)
 161  */
 162 void smpi_process_mark_as_initialized(void)
 163 {
 164   int index = smpi_process_index();
 165   if ((index != MPI_UNDEFINED) && (!process_data[index]->state != SMPI_FINALIZED))
 166     process_data[index]->state = SMPI_INITIALIZED;
 167 }
 168
 169
 170 #ifdef SMPI_F2C
 171 int smpi_process_argc(void)
 172 {
 173   smpi_process_data_t data = smpi_process_data();
 174   return data->argc ? *(data->argc) - 1 : 0;
 175 }
 176
 177 int smpi_process_getarg(integer * index, char *dst, ftnlen len)
 178 {
 179   smpi_process_data_t data = smpi_process_data();
 180   char *arg;
 181   ftnlen i;
 182
 183   if (!data->argc || !data->argv || *index < 1 || *index >= *(data->argc)) {
 184     return -1;
 185   }
 186   arg = (*data->argv)[*index];
 187   for (i = 0; i < len && arg[i] != '\0'; i++) {
 188     dst[i] = arg[i];
 189   }
 190   for (; i < len; i++) {
 191     dst[i] = ' ';
 192   }
 193   return 0;
 194 }
 195 #endif
 196
 197 int smpi_global_size(void)
 198 {
 199   char *value = getenv("SMPI_GLOBAL_SIZE");
 200
 201   if (!value) {
 202     fprintf(stderr,
 203             "Please set env var SMPI_GLOBAL_SIZE to expected number of processes.\n");
 204     xbt_abort();
 205   }
 206   return atoi(value);
 207 }
 208
 209 smpi_process_data_t smpi_process_data(void)
 210 {
 211   return SIMIX_process_self_get_data(SIMIX_process_self());
 212 }
 213
 214 smpi_process_data_t smpi_process_remote_data(int index)
 215 {
 216   return process_data[index];
 217 }
 218
 219 void smpi_process_set_user_data(void *data)
 220 {
 221   smpi_process_data_t process_data = smpi_process_data();
 222   process_data->data = data;
 223 }
 224
 225 void *smpi_process_get_user_data()
 226 {
 227   smpi_process_data_t process_data = smpi_process_data();
 228   return process_data->data;
 229 }
 230
 231 int smpi_process_count(void)
 232 {
 233   return process_count;
 234 }
 235
 236 int smpi_process_index(void)
 237 {
 238   smpi_process_data_t data = smpi_process_data();
 239   //return -1 if not initialized
 240   return data ? data->index : MPI_UNDEFINED;
 241 }
 242
 243 smx_rdv_t smpi_process_mailbox(void)
 244 {
 245   smpi_process_data_t data = smpi_process_data();
 246   return data->mailbox;
 247 }
 248
 249 smx_rdv_t smpi_process_mailbox_small(void)
 250 {
 251   smpi_process_data_t data = smpi_process_data();
 252   return data->mailbox_small;
 253 }
 254
 255 smx_rdv_t smpi_process_remote_mailbox(int index)
 256 {
 257   smpi_process_data_t data = smpi_process_remote_data(index);
 258   return data->mailbox;
 259 }
 260
 261
 262 smx_rdv_t smpi_process_remote_mailbox_small(int index)
 263 {
 264   smpi_process_data_t data = smpi_process_remote_data(index);
 265   return data->mailbox_small;
 266 }
 267
 268 xbt_os_timer_t smpi_process_timer(void)
 269 {
 270   smpi_process_data_t data = smpi_process_data();
 271   return data->timer;
 272 }
 273
 274 void smpi_process_simulated_start(void)
 275 {
 276   smpi_process_data_t data = smpi_process_data();
 277   data->simulated = SIMIX_get_clock();
 278 }
 279
 280 double smpi_process_simulated_elapsed(void)
 281 {
 282   smpi_process_data_t data = smpi_process_data();
 283   return SIMIX_get_clock() - data->simulated;
 284 }
 285
 286 MPI_Comm smpi_process_comm_self(void)
 287 {
 288   smpi_process_data_t data = smpi_process_data();
 289   if(data->comm_self==MPI_COMM_NULL){
 290     MPI_Group group = smpi_group_new(1);
 291     data->comm_self = smpi_comm_new(group);
 292     smpi_group_set_mapping(group, smpi_process_index(), 0);
 293   }
 294
 295   return data->comm_self;
 296 }
 297
 298 void smpi_process_set_sampling(int s)
 299 {
 300   smpi_process_data_t data = smpi_process_data();
 301   data->sampling = s;
 302 }
 303
 304 int smpi_process_get_sampling(void)
 305 {
 306   smpi_process_data_t data = smpi_process_data();
 307   return data->sampling;
 308 }
 309
 310 void print_request(const char *message, MPI_Request request)
 311 {
 312   XBT_DEBUG
 313       ("%s  request %p  [buf = %p, size = %zu, src = %d, dst = %d, tag = %d, flags = %x]",
 314        message, request, request->buf, request->size, request->src,
 315        request->dst, request->tag, request->flags);
 316 }
 317
 318 static void smpi_comm_copy_buffer_callback(smx_action_t comm,
 319                                            void *buff, size_t buff_size)
 320 {
 321   XBT_DEBUG("Copy the data over");
 322   if(_xbt_replay_is_active()) return;
 323   memcpy(comm->comm.dst_buff, buff, buff_size);
 324   if (comm->comm.detached) {
 325     // if this is a detached send, the source buffer was duplicated by SMPI
 326     // sender to make the original buffer available to the application ASAP
 327     xbt_free(buff);
 328     //It seems that the request is used after the call there this should
 329     //be free somewhereelse  but where???
 330     //xbt_free(comm->comm.src_data);// inside SMPI the request is keep
 331     //inside the user data and should be free
 332     comm->comm.src_buff = NULL;
 333   }
 334 }
 335
 336 void smpi_global_init(void)
 337 {
 338   int i;
 339   MPI_Group group;
 340   char name[MAILBOX_NAME_MAXLEN];
 341
 342   SIMIX_comm_set_copy_data_callback(&smpi_comm_copy_buffer_callback);
 343   process_count = SIMIX_process_count();
 344   process_data = xbt_new(smpi_process_data_t, process_count);
 345   for (i = 0; i < process_count; i++) {
 346     process_data[i] = xbt_new(s_smpi_process_data_t, 1);
 347     process_data[i]->index = i;
 348     process_data[i]->argc = NULL;
 349     process_data[i]->argv = NULL;
 350     process_data[i]->mailbox = simcall_rdv_create(get_mailbox_name(name, i));
 351     process_data[i]->mailbox_small =
 352         simcall_rdv_create(get_mailbox_name_small(name, i));
 353     process_data[i]->timer = xbt_os_timer_new();
 354     if (MC_is_active())
 355       MC_ignore_heap(process_data[i]->timer, xbt_os_timer_size());
 356     process_data[i]->comm_self = MPI_COMM_NULL;
 357     process_data[i]->state = SMPI_UNINITIALIZED;
 358     process_data[i]->sampling = 0;
 359   }
 360   group = smpi_group_new(process_count);
 361   MPI_COMM_WORLD = smpi_comm_new(group);
 362   MPI_UNIVERSE_SIZE = smpi_comm_size(MPI_COMM_WORLD);
 363   for (i = 0; i < process_count; i++) {
 364     smpi_group_set_mapping(group, i, i);
 365   }
 366
 367   //check correctness of MPI parameters
 368
 369   xbt_assert(sg_cfg_get_int("smpi/async_small_thres") <=
 370              sg_cfg_get_int("smpi/send_is_detached_thres"));
 371
 372   if (sg_cfg_is_default_value("smpi/running_power")) {
 373     XBT_INFO("You did not set the power of the host running the simulation.  "
 374              "The timings will certainly not be accurate.  "
 375              "Use the option \"--cfg=smpi/running_power:<flops>\" to set its value."
 376              "Check http://simgrid.org/simgrid/latest/doc/options.html#options_smpi_bench for more information. ");
 377   }
 378 }
 379
 380 void smpi_global_destroy(void)
 381 {
 382   int count = smpi_process_count();
 383   int i;
 384
 385   smpi_bench_destroy();
 386   while (smpi_group_unuse(smpi_comm_group(MPI_COMM_WORLD)) > 0);
 387   xbt_free(MPI_COMM_WORLD);
 388   MPI_COMM_WORLD = MPI_COMM_NULL;
 389   for (i = 0; i < count; i++) {
 390     if(process_data[i]->comm_self!=MPI_COMM_NULL){
 391       smpi_group_unuse(smpi_comm_group(process_data[i]->comm_self));
 392       smpi_comm_destroy(process_data[i]->comm_self);
 393     }
 394     xbt_os_timer_free(process_data[i]->timer);
 395     simcall_rdv_destroy(process_data[i]->mailbox);
 396     simcall_rdv_destroy(process_data[i]->mailbox_small);
 397     xbt_free(process_data[i]);
 398   }
 399   xbt_free(process_data);
 400   process_data = NULL;
 401
 402   smpi_free_static();
 403 }
 404
 405 /* Fortran specific stuff */
 406 /* With smpicc, the following weak symbols are used */
 407 /* With smpiff, the following weak symbols are replaced by those in libf2c */
 408 int __attribute__ ((weak)) xargc;
 409 char ** __attribute__ ((weak)) xargv;
 410
 411 #ifndef WIN32
 412 void __attribute__ ((weak)) user_main_()
 413 {
 414   xbt_die("Should not be in this smpi_simulated_main");
 415   return;
 416 }
 417
 418 int __attribute__ ((weak)) smpi_simulated_main_(int argc, char **argv)
 419 {
 420   smpi_process_init(&argc, &argv);
 421   user_main_();
 422   return 0;
 423 }
 424
 425 int __attribute__ ((weak)) main(int argc, char **argv)
 426 {
 427   return smpi_main(smpi_simulated_main_, argc, argv);
 428 }
 429
 430 int __attribute__ ((weak)) MAIN__()
 431 {
 432   return smpi_main(smpi_simulated_main_, xargc, xargv);
 433 };
 434 #endif
 435
 436 int smpi_main(int (*realmain) (int argc, char *argv[]), int argc, char *argv[])
 437 {
 438   srand(SMPI_RAND_SEED);
 439
 440   if (getenv("SMPI_PRETEND_CC") != NULL) {
 441     /* Hack to ensure that smpicc can pretend to be a simple
 442      * compiler. Particularly handy to pass it to the configuration tools */
 443     return 0;
 444   }
 445
 446   /* Connect log categories.  See xbt/log.c */
 447   XBT_LOG_CONNECT(smpi);        /* Keep this line as soon as possible in this
 448                                    function: xbt_log_appender_file.c depends on it
 449                                    DO NOT connect this in XBT or so, or it will be
 450                                    useless to xbt_log_appender_file.c */
 451 #ifdef HAVE_TRACING
 452   XBT_LOG_CONNECT(instr_smpi);
 453 #endif
 454   XBT_LOG_CONNECT(smpi_base);
 455   XBT_LOG_CONNECT(smpi_bench);
 456   XBT_LOG_CONNECT(smpi_coll);
 457   XBT_LOG_CONNECT(smpi_colls);
 458   XBT_LOG_CONNECT(smpi_comm);
 459   XBT_LOG_CONNECT(smpi_dvfs);
 460   XBT_LOG_CONNECT(smpi_group);
 461   XBT_LOG_CONNECT(smpi_kernel);
 462   XBT_LOG_CONNECT(smpi_mpi);
 463   XBT_LOG_CONNECT(smpi_mpi_dt);
 464   XBT_LOG_CONNECT(smpi_pmpi);
 465   XBT_LOG_CONNECT(smpi_replay);
 466
 467 #ifdef HAVE_TRACING
 468   TRACE_global_init(&argc, argv);
 469
 470   TRACE_add_start_function(TRACE_smpi_alloc);
 471   TRACE_add_end_function(TRACE_smpi_release);
 472 #endif
 473
 474   SIMIX_global_init(&argc, argv);
 475
 476 #ifdef HAVE_TRACING
 477   TRACE_start();
 478 #endif
 479
 480   // parse the platform file: get the host list
 481   SIMIX_create_environment(argv[1]);
 482
 483   SIMIX_function_register_default(realmain);
 484   SIMIX_launch_application(argv[2]);
 485
 486   int gather_id = find_coll_description(mpi_coll_gather_description,
 487                                         sg_cfg_get_string("smpi/gather"));
 488   mpi_coll_gather_fun = (int (*)(void *, int, MPI_Datatype,
 489                                  void *, int, MPI_Datatype, int, MPI_Comm))
 490       mpi_coll_gather_description[gather_id].coll;
 491
 492   int allgather_id = find_coll_description(mpi_coll_allgather_description,
 493                                            sg_cfg_get_string("smpi/allgather"));
 494   mpi_coll_allgather_fun = (int (*)(void *, int, MPI_Datatype,
 495                                     void *, int, MPI_Datatype, MPI_Comm))
 496       mpi_coll_allgather_description[allgather_id].coll;
 497
 498   int allgatherv_id = find_coll_description(mpi_coll_allgatherv_description,
 499                                             sg_cfg_get_string("smpi/allgatherv"));
 500   mpi_coll_allgatherv_fun = (int (*)(void *, int, MPI_Datatype, void *, int *,
 501                                      int *, MPI_Datatype, MPI_Comm))
 502       mpi_coll_allgatherv_description[allgatherv_id].coll;
 503
 504   int allreduce_id = find_coll_description(mpi_coll_allreduce_description,
 505                                            sg_cfg_get_string("smpi/allreduce"));
 506   mpi_coll_allreduce_fun = (int (*)(void *sbuf, void *rbuf, int rcount,
 507                                     MPI_Datatype dtype, MPI_Op op,
 508                                     MPI_Comm comm))
 509       mpi_coll_allreduce_description[allreduce_id].coll;
 510
 511   int alltoall_id = find_coll_description(mpi_coll_alltoall_description,
 512                                           sg_cfg_get_string("smpi/alltoall"));
 513   mpi_coll_alltoall_fun = (int (*)(void *, int, MPI_Datatype,
 514                                    void *, int, MPI_Datatype, MPI_Comm))
 515       mpi_coll_alltoall_description[alltoall_id].coll;
 516
 517   int alltoallv_id = find_coll_description(mpi_coll_alltoallv_description,
 518                                            sg_cfg_get_string("smpi/alltoallv"));
 519   mpi_coll_alltoallv_fun = (int (*)(void *, int *, int *, MPI_Datatype,
 520                                     void *, int *, int *, MPI_Datatype,
 521                                     MPI_Comm))
 522       mpi_coll_alltoallv_description[alltoallv_id].coll;
 523
 524   int bcast_id = find_coll_description(mpi_coll_bcast_description,
 525                                        sg_cfg_get_string("smpi/bcast"));
 526   mpi_coll_bcast_fun = (int (*)(void *buf, int count, MPI_Datatype datatype,
 527                                 int root, MPI_Comm com))
 528       mpi_coll_bcast_description[bcast_id].coll;
 529
 530   int reduce_id = find_coll_description(mpi_coll_reduce_description,
 531                                         sg_cfg_get_string("smpi/reduce"));
 532   mpi_coll_reduce_fun = (int (*)(void *buf, void *rbuf, int count,
 533                                  MPI_Datatype datatype, MPI_Op op,
 534                                  int root, MPI_Comm comm))
 535       mpi_coll_reduce_description[reduce_id].coll;
 536
 537   int reduce_scatter_id =
 538       find_coll_description(mpi_coll_reduce_scatter_description,
 539                             sg_cfg_get_string("smpi/reduce_scatter"));
 540   mpi_coll_reduce_scatter_fun = (int (*)(void *sbuf, void *rbuf, int *rcounts,
 541                                          MPI_Datatype dtype, MPI_Op op,
 542                                          MPI_Comm comm))
 543       mpi_coll_reduce_scatter_description[reduce_scatter_id].coll;
 544
 545   int scatter_id = find_coll_description(mpi_coll_scatter_description,
 546                                          sg_cfg_get_string("smpi/scatter"));
 547   mpi_coll_scatter_fun = (int (*)(void *sendbuf, int sendcount,
 548                                   MPI_Datatype sendtype, void *recvbuf,
 549                                   int recvcount, MPI_Datatype recvtype,
 550                                   int root, MPI_Comm comm))
 551       mpi_coll_scatter_description[scatter_id].coll;
 552
 553   int barrier_id = find_coll_description(mpi_coll_barrier_description,
 554                                          sg_cfg_get_string("smpi/barrier"));
 555   mpi_coll_barrier_fun = (int (*)(MPI_Comm comm))
 556       mpi_coll_barrier_description[barrier_id].coll;
 557
 558   smpi_cpu_threshold = sg_cfg_get_double("smpi/cpu_threshold");
 559   smpi_running_power = sg_cfg_get_double("smpi/running_power");
 560   if (smpi_cpu_threshold < 0)
 561     smpi_cpu_threshold = DBL_MAX;
 562
 563   smpi_global_init();
 564
 565   /* Clean IO before the run */
 566   fflush(stdout);
 567   fflush(stderr);
 568
 569   if (MC_is_active())
 570     MC_do_the_modelcheck_for_real();
 571   else
 572     SIMIX_run();
 573
 574   if (sg_cfg_get_boolean("smpi/display_timing"))
 575     XBT_INFO("Simulation time: %g seconds.", SIMIX_get_clock());
 576
 577   smpi_global_destroy();
 578
 579 #ifdef HAVE_TRACING
 580   TRACE_end();
 581 #endif
 582
 583   return 0;
 584 }