src/smpi/smpi_global.c

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