src/smpi/smpi_replay.c

   1 /* Copyright (c) 2009-2013. 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
   8 #include "private.h"
   9 #include <stdio.h>
  10 #include <xbt.h>
  11 #include <xbt/replay.h>
  12
  13 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_replay,smpi,"Trace Replay with SMPI");
  14
  15 int communicator_size = 0;
  16 static int active_processes = 0;
  17 xbt_dynar_t *reqq = NULL;
  18
  19 MPI_Datatype MPI_DEFAULT_TYPE;
  20 MPI_Datatype MPI_CURRENT_TYPE;
  21
  22 static void log_timed_action (const char *const *action, double clock){
  23   if (XBT_LOG_ISENABLED(smpi_replay, xbt_log_priority_verbose)){
  24     char *name = xbt_str_join_array(action, " ");
  25     XBT_VERB("%s %f", name, smpi_process_simulated_elapsed()-clock);
  26     free(name);
  27   }
  28 }
  29
  30 typedef struct {
  31   xbt_dynar_t irecvs; /* of MPI_Request */
  32 } s_smpi_replay_globals_t, *smpi_replay_globals_t;
  33
  34
  35 /* Helper function */
  36 static double parse_double(const char *string)
  37 {
  38   double value;
  39   char *endptr;
  40   value = strtod(string, &endptr);
  41   if (*endptr != '\0')
  42     THROWF(unknown_error, 0, "%s is not a double", string);
  43   return value;
  44 }
  45
  46 static MPI_Datatype decode_datatype(const char *const action)
  47 {
  48 // Declared datatypes,
  49
  50   switch(atoi(action))
  51   {
  52     case 0:
  53       MPI_CURRENT_TYPE=MPI_DOUBLE;
  54       break;
  55     case 1:
  56       MPI_CURRENT_TYPE=MPI_INT;
  57       break;
  58     case 2:
  59       MPI_CURRENT_TYPE=MPI_CHAR;
  60       break;
  61     case 3:
  62       MPI_CURRENT_TYPE=MPI_SHORT;
  63       break;
  64     case 4:
  65       MPI_CURRENT_TYPE=MPI_LONG;
  66       break;
  67     case 5:
  68       MPI_CURRENT_TYPE=MPI_FLOAT;
  69       break;
  70     case 6:
  71       MPI_CURRENT_TYPE=MPI_BYTE;
  72       break;
  73     default:
  74       MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
  75
  76   }
  77    return MPI_CURRENT_TYPE;
  78 }
  79
  80
  81 const char* encode_datatype(MPI_Datatype datatype)
  82 {
  83
  84   //default type for output is set to MPI_BYTE
  85   // MPI_DEFAULT_TYPE is not set for output, use directly MPI_BYTE
  86   if (datatype==MPI_BYTE){
  87       return "";
  88   }
  89   if(datatype==MPI_DOUBLE)
  90       return "0";
  91   if(datatype==MPI_INT)
  92       return "1";
  93   if(datatype==MPI_CHAR)
  94       return "2";
  95   if(datatype==MPI_SHORT)
  96       return "3";
  97   if(datatype==MPI_LONG)
  98     return "4";
  99   if(datatype==MPI_FLOAT)
 100       return "5";
 101
 102   // default - not implemented.
 103   // do not warn here as we pass in this function even for other trace formats
 104   return "-1";
 105 }
 106
 107 static void action_init(const char *const *action)
 108 {
 109   int i;
 110   XBT_DEBUG("Initialize the counters");
 111   smpi_replay_globals_t globals =  xbt_new(s_smpi_replay_globals_t, 1);
 112   globals->irecvs = xbt_dynar_new(sizeof(MPI_Request),NULL);
 113
 114   if(action[2]) MPI_DEFAULT_TYPE= MPI_DOUBLE; // default MPE dataype
 115   else MPI_DEFAULT_TYPE= MPI_BYTE; // default TAU datatype
 116
 117   smpi_process_set_user_data((void*) globals);
 118
 119   /* start a simulated timer */
 120   smpi_process_simulated_start();
 121   /*initialize the number of active processes */
 122   active_processes = smpi_process_count();
 123
 124   if (!reqq) {
 125     reqq=xbt_new0(xbt_dynar_t,active_processes);
 126
 127     for(i=0;i<active_processes;i++){
 128       reqq[i]=xbt_dynar_new(sizeof(MPI_Request),NULL);
 129     }
 130   }
 131 }
 132
 133 static void action_finalize(const char *const *action)
 134 {
 135   smpi_replay_globals_t globals =
 136       (smpi_replay_globals_t) smpi_process_get_user_data();
 137   if (globals){
 138     XBT_DEBUG("There are %lu irecvs in the dynar",
 139          xbt_dynar_length(globals->irecvs));
 140     xbt_dynar_free_container(&(globals->irecvs));
 141   }
 142   free(globals);
 143 }
 144
 145 static void action_comm_size(const char *const *action)
 146 {
 147   double clock = smpi_process_simulated_elapsed();
 148
 149   communicator_size = parse_double(action[2]);
 150   log_timed_action (action, clock);
 151 }
 152
 153 static void action_comm_split(const char *const *action)
 154 {
 155   double clock = smpi_process_simulated_elapsed();
 156
 157   log_timed_action (action, clock);
 158 }
 159
 160 static void action_comm_dup(const char *const *action)
 161 {
 162   double clock = smpi_process_simulated_elapsed();
 163
 164   log_timed_action (action, clock);
 165 }
 166
 167 static void action_compute(const char *const *action)
 168 {
 169   double clock = smpi_process_simulated_elapsed();
 170   double flops= parse_double(action[2]);
 171 #ifdef HAVE_TRACING
 172   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 173   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 174   extra->type=TRACING_COMPUTING;
 175   extra->comp_size=flops;
 176   TRACE_smpi_computing_in(rank, extra);
 177 #endif
 178   smpi_execute_flops(flops);
 179 #ifdef HAVE_TRACING
 180   TRACE_smpi_computing_out(rank);
 181 #endif
 182
 183   log_timed_action (action, clock);
 184 }
 185
 186 static void action_send(const char *const *action)
 187 {
 188   int to = atoi(action[2]);
 189   double size=parse_double(action[3]);
 190   double clock = smpi_process_simulated_elapsed();
 191
 192   if(action[4]) {
 193     MPI_CURRENT_TYPE=decode_datatype(action[4]);
 194   } else {
 195     MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 196   }
 197
 198 #ifdef HAVE_TRACING
 199   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 200
 201   int dst_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), to);
 202   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 203   extra->type = TRACING_SEND;
 204   extra->send_size = size;
 205   extra->src = rank;
 206   extra->dst = dst_traced;
 207   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 208   TRACE_smpi_ptp_in(rank, rank, dst_traced, __FUNCTION__, extra);
 209   TRACE_smpi_send(rank, rank, dst_traced, size*smpi_datatype_size(MPI_CURRENT_TYPE));
 210 #endif
 211
 212   smpi_mpi_send(NULL, size, MPI_CURRENT_TYPE, to , 0, MPI_COMM_WORLD);
 213
 214   log_timed_action (action, clock);
 215
 216   #ifdef HAVE_TRACING
 217   TRACE_smpi_ptp_out(rank, rank, dst_traced, __FUNCTION__);
 218 #endif
 219
 220 }
 221
 222 static void action_Isend(const char *const *action)
 223 {
 224   int to = atoi(action[2]);
 225   double size=parse_double(action[3]);
 226   double clock = smpi_process_simulated_elapsed();
 227   MPI_Request request;
 228
 229   if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
 230   else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 231
 232 #ifdef HAVE_TRACING
 233   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 234   int dst_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), to);
 235   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 236   extra->type = TRACING_ISEND;
 237   extra->send_size = size;
 238   extra->src = rank;
 239   extra->dst = dst_traced;
 240   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 241   TRACE_smpi_ptp_in(rank, rank, dst_traced, __FUNCTION__, extra);
 242   TRACE_smpi_send(rank, rank, dst_traced, size*smpi_datatype_size(MPI_CURRENT_TYPE));
 243 #endif
 244
 245   request = smpi_mpi_isend(NULL, size, MPI_CURRENT_TYPE, to, 0,MPI_COMM_WORLD);
 246
 247 #ifdef HAVE_TRACING
 248   TRACE_smpi_ptp_out(rank, rank, dst_traced, __FUNCTION__);
 249   request->send = 1;
 250 #endif
 251
 252   xbt_dynar_push(reqq[smpi_comm_rank(MPI_COMM_WORLD)],&request);
 253
 254   log_timed_action (action, clock);
 255 }
 256
 257 static void action_recv(const char *const *action) {
 258   int from = atoi(action[2]);
 259   double size=parse_double(action[3]);
 260   double clock = smpi_process_simulated_elapsed();
 261   MPI_Status status;
 262
 263   if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
 264   else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 265
 266 #ifdef HAVE_TRACING
 267   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 268   int src_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), from);
 269
 270   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 271   extra->type = TRACING_RECV;
 272   extra->send_size = size;
 273   extra->src = src_traced;
 274   extra->dst = rank;
 275   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 276   TRACE_smpi_ptp_in(rank, src_traced, rank, __FUNCTION__, extra);
 277 #endif
 278
 279   smpi_mpi_recv(NULL, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD, &status);
 280
 281 #ifdef HAVE_TRACING
 282   TRACE_smpi_ptp_out(rank, src_traced, rank, __FUNCTION__);
 283   TRACE_smpi_recv(rank, src_traced, rank);
 284 #endif
 285
 286   log_timed_action (action, clock);
 287 }
 288
 289 static void action_Irecv(const char *const *action)
 290 {
 291   int from = atoi(action[2]);
 292   double size=parse_double(action[3]);
 293   double clock = smpi_process_simulated_elapsed();
 294   MPI_Request request;
 295
 296   smpi_replay_globals_t globals =
 297      (smpi_replay_globals_t) smpi_process_get_user_data();
 298
 299   if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
 300   else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 301
 302 #ifdef HAVE_TRACING
 303   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 304   int src_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), from);
 305   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 306   extra->type = TRACING_IRECV;
 307   extra->send_size = size;
 308   extra->src = src_traced;
 309   extra->dst = rank;
 310   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 311   TRACE_smpi_ptp_in(rank, src_traced, rank, __FUNCTION__, extra);
 312 #endif
 313
 314   request = smpi_mpi_irecv(NULL, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD);
 315
 316 #ifdef HAVE_TRACING
 317   TRACE_smpi_ptp_out(rank, src_traced, rank, __FUNCTION__);
 318   request->recv = 1;
 319 #endif
 320   xbt_dynar_push(globals->irecvs,&request);
 321   xbt_dynar_push(reqq[smpi_comm_rank(MPI_COMM_WORLD)],&request);
 322
 323   log_timed_action (action, clock);
 324 }
 325
 326 static void action_wait(const char *const *action){
 327   double clock = smpi_process_simulated_elapsed();
 328   MPI_Request request;
 329   MPI_Status status;
 330   smpi_replay_globals_t globals =
 331       (smpi_replay_globals_t) smpi_process_get_user_data();
 332
 333   xbt_assert(xbt_dynar_length(globals->irecvs),
 334       "action wait not preceded by any irecv: %s",
 335       xbt_str_join_array(action," "));
 336   request = xbt_dynar_pop_as(globals->irecvs,MPI_Request);
 337 #ifdef HAVE_TRACING
 338   int rank = request && request->comm != MPI_COMM_NULL
 339       ? smpi_comm_rank(request->comm)
 340       : -1;
 341
 342   MPI_Group group = smpi_comm_group(request->comm);
 343   int src_traced = smpi_group_rank(group, request->src);
 344   int dst_traced = smpi_group_rank(group, request->dst);
 345   int is_wait_for_receive = request->recv;
 346   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 347   extra->type = TRACING_WAIT;
 348   TRACE_smpi_ptp_in(rank, src_traced, dst_traced, __FUNCTION__, extra);
 349 #endif
 350   smpi_mpi_wait(&request, &status);
 351 #ifdef HAVE_TRACING
 352   TRACE_smpi_ptp_out(rank, src_traced, dst_traced, __FUNCTION__);
 353   if (is_wait_for_receive) {
 354     TRACE_smpi_recv(rank, src_traced, dst_traced);
 355   }
 356 #endif
 357
 358   log_timed_action (action, clock);
 359 }
 360
 361 static void action_waitall(const char *const *action){
 362   double clock = smpi_process_simulated_elapsed();
 363   int count_requests=0;
 364   unsigned int i=0;
 365
 366   count_requests=xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
 367
 368   if (count_requests>0) {
 369     MPI_Request requests[count_requests];
 370     MPI_Status status[count_requests];
 371
 372     /*  The reqq is an array of dynars. Its index corresponds to the rank.
 373      Thus each rank saves its own requests to the array request. */
 374     xbt_dynar_foreach(reqq[smpi_comm_rank(MPI_COMM_WORLD)],i,requests[i]);
 375
 376   #ifdef HAVE_TRACING
 377    //save information from requests
 378
 379    xbt_dynar_t srcs = xbt_dynar_new(sizeof(int), NULL);
 380    xbt_dynar_t dsts = xbt_dynar_new(sizeof(int), NULL);
 381    xbt_dynar_t recvs = xbt_dynar_new(sizeof(int), NULL);
 382    for (i = 0; i < count_requests; i++) {
 383     if(requests[i]){
 384       int *asrc = xbt_new(int, 1);
 385       int *adst = xbt_new(int, 1);
 386       int *arecv = xbt_new(int, 1);
 387       *asrc = requests[i]->src;
 388       *adst = requests[i]->dst;
 389       *arecv = requests[i]->recv;
 390       xbt_dynar_insert_at(srcs, i, asrc);
 391       xbt_dynar_insert_at(dsts, i, adst);
 392       xbt_dynar_insert_at(recvs, i, arecv);
 393       xbt_free(asrc);
 394       xbt_free(adst);
 395       xbt_free(arecv);
 396     }else {
 397       int *t = xbt_new(int, 1);
 398       xbt_dynar_insert_at(srcs, i, t);
 399       xbt_dynar_insert_at(dsts, i, t);
 400       xbt_dynar_insert_at(recvs, i, t);
 401       xbt_free(t);
 402     }
 403    }
 404    int rank_traced = smpi_process_index();
 405    instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 406    extra->type = TRACING_WAITALL;
 407    extra->send_size=count_requests;
 408    TRACE_smpi_ptp_in(rank_traced, -1, -1, __FUNCTION__,extra);
 409  #endif
 410
 411     smpi_mpi_waitall(count_requests, requests, status);
 412
 413   #ifdef HAVE_TRACING
 414    for (i = 0; i < count_requests; i++) {
 415     int src_traced, dst_traced, is_wait_for_receive;
 416     xbt_dynar_get_cpy(srcs, i, &src_traced);
 417     xbt_dynar_get_cpy(dsts, i, &dst_traced);
 418     xbt_dynar_get_cpy(recvs, i, &is_wait_for_receive);
 419     if (is_wait_for_receive) {
 420       TRACE_smpi_recv(rank_traced, src_traced, dst_traced);
 421     }
 422    }
 423    TRACE_smpi_ptp_out(rank_traced, -1, -1, __FUNCTION__);
 424    //clean-up of dynars
 425    xbt_dynar_free(&srcs);
 426    xbt_dynar_free(&dsts);
 427    xbt_dynar_free(&recvs);
 428   #endif
 429
 430    xbt_dynar_reset(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
 431   }
 432   log_timed_action (action, clock);
 433 }
 434
 435 static void action_barrier(const char *const *action){
 436   double clock = smpi_process_simulated_elapsed();
 437 #ifdef HAVE_TRACING
 438   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 439   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 440   extra->type = TRACING_BARRIER;
 441   TRACE_smpi_collective_in(rank, -1, __FUNCTION__, extra);
 442 #endif
 443   smpi_mpi_barrier(MPI_COMM_WORLD);
 444 #ifdef HAVE_TRACING
 445   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 446 #endif
 447
 448   log_timed_action (action, clock);
 449 }
 450
 451
 452 static void action_bcast(const char *const *action)
 453 {
 454   double size = parse_double(action[2]);
 455   double clock = smpi_process_simulated_elapsed();
 456   int root=0;
 457   /*
 458    * Initialize MPI_CURRENT_TYPE in order to decrease
 459    * the number of the checks
 460    * */
 461   MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 462
 463   if(action[3]) {
 464     root= atoi(action[3]);
 465     if(action[4]) {
 466       MPI_CURRENT_TYPE=decode_datatype(action[4]);
 467     }
 468   }
 469
 470 #ifdef HAVE_TRACING
 471   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 472   int root_traced = smpi_group_index(smpi_comm_group(MPI_COMM_WORLD), root);
 473
 474   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 475   extra->type = TRACING_BCAST;
 476   extra->send_size = size;
 477   extra->root = root_traced;
 478   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 479   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__, extra);
 480
 481 #endif
 482
 483   mpi_coll_bcast_fun(NULL, size, MPI_CURRENT_TYPE, root, MPI_COMM_WORLD);
 484 #ifdef HAVE_TRACING
 485   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
 486 #endif
 487
 488   log_timed_action (action, clock);
 489 }
 490
 491 static void action_reduce(const char *const *action)
 492 {
 493   double comm_size = parse_double(action[2]);
 494   double comp_size = parse_double(action[3]);
 495   double clock = smpi_process_simulated_elapsed();
 496   int root=0;
 497   MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 498
 499   if(action[4]) {
 500     root= atoi(action[4]);
 501     if(action[5]) {
 502       MPI_CURRENT_TYPE=decode_datatype(action[5]);
 503     }
 504   }
 505
 506 #ifdef HAVE_TRACING
 507   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 508   int root_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), root);
 509   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 510   extra->type = TRACING_REDUCE;
 511   extra->send_size = comm_size;
 512   extra->comp_size = comp_size;
 513   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 514   extra->root = root_traced;
 515
 516   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__,extra);
 517 #endif
 518    mpi_coll_reduce_fun(NULL, NULL, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, root, MPI_COMM_WORLD);
 519    smpi_execute_flops(comp_size);
 520 #ifdef HAVE_TRACING
 521   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
 522 #endif
 523
 524   log_timed_action (action, clock);
 525 }
 526
 527 static void action_allReduce(const char *const *action) {
 528   double comm_size = parse_double(action[2]);
 529   double comp_size = parse_double(action[3]);
 530
 531   if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
 532   else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 533
 534   double clock = smpi_process_simulated_elapsed();
 535 #ifdef HAVE_TRACING
 536   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 537   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 538   extra->type = TRACING_ALLREDUCE;
 539   extra->send_size = comm_size;
 540   extra->comp_size = comp_size;
 541   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 542
 543   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 544 #endif
 545   mpi_coll_reduce_fun(NULL, NULL, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, 0, MPI_COMM_WORLD);
 546   smpi_execute_flops(comp_size);
 547   mpi_coll_bcast_fun(NULL, comm_size, MPI_CURRENT_TYPE, 0, MPI_COMM_WORLD);
 548 #ifdef HAVE_TRACING
 549   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 550 #endif
 551
 552   log_timed_action (action, clock);
 553 }
 554
 555 static void action_allToAll(const char *const *action) {
 556   double clock = smpi_process_simulated_elapsed();
 557   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 558   int send_size = parse_double(action[2]);
 559   int recv_size = parse_double(action[3]);
 560   MPI_Datatype MPI_CURRENT_TYPE2;
 561
 562   if(action[4]) {
 563     MPI_CURRENT_TYPE=decode_datatype(action[4]);
 564     MPI_CURRENT_TYPE2=decode_datatype(action[5]);
 565   }
 566   else {
 567     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 568     MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
 569   }
 570   void *send = calloc(send_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE));
 571   void *recv = calloc(recv_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
 572
 573 #ifdef HAVE_TRACING
 574   int rank = smpi_process_index();
 575   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 576   extra->type = TRACING_ALLTOALL;
 577   extra->send_size = send_size;
 578   extra->recv_size = recv_size;
 579   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 580   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
 581
 582   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 583 #endif
 584
 585   mpi_coll_alltoall_fun(send, send_size, MPI_CURRENT_TYPE, recv, recv_size, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
 586
 587 #ifdef HAVE_TRACING
 588   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 589 #endif
 590
 591   log_timed_action (action, clock);
 592   xbt_free(send);
 593   xbt_free(recv);
 594 }
 595
 596
 597 static void action_gather(const char *const *action) {
 598   /*
 599  The structure of the gather action for the rank 0 (total 4 processes)
 600  is the following:
 601  0 gather 68 68 0 0 0
 602
 603   where:
 604   1) 68 is the sendcounts
 605   2) 68 is the recvcounts
 606   3) 0 is the root node
 607   4) 0 is the send datatype id, see decode_datatype()
 608   5) 0 is the recv datatype id, see decode_datatype()
 609   */
 610   double clock = smpi_process_simulated_elapsed();
 611   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 612   int send_size = parse_double(action[2]);
 613   int recv_size = parse_double(action[3]);
 614   MPI_Datatype MPI_CURRENT_TYPE2;
 615   if(action[5]) {
 616     MPI_CURRENT_TYPE=decode_datatype(action[5]);
 617     MPI_CURRENT_TYPE2=decode_datatype(action[6]);
 618   } else {
 619     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 620     MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
 621   }
 622   void *send = calloc(send_size, smpi_datatype_size(MPI_CURRENT_TYPE));
 623   void *recv = NULL;
 624
 625   int root=atoi(action[4]);
 626   int rank = smpi_process_index();
 627
 628   if(rank==root)
 629     recv = calloc(recv_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
 630
 631 #ifdef HAVE_TRACING
 632   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 633   extra->type = TRACING_GATHER;
 634   extra->send_size = send_size;
 635   extra->recv_size = recv_size;
 636   extra->root = root;
 637   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 638   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
 639
 640   TRACE_smpi_collective_in(rank, root, __FUNCTION__, extra);
 641 #endif
 642 smpi_mpi_gather(send, send_size, MPI_CURRENT_TYPE,
 643                 recv, recv_size, MPI_CURRENT_TYPE2,
 644                 root, MPI_COMM_WORLD);
 645
 646 #ifdef HAVE_TRACING
 647   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 648 #endif
 649
 650   log_timed_action (action, clock);
 651   xbt_free(send);
 652   xbt_free(recv);
 653 }
 654
 655
 656
 657 static void action_gatherv(const char *const *action) {
 658   /*
 659  The structure of the gatherv action for the rank 0 (total 4 processes)
 660  is the following:
 661  0 gather 68 68 10 10 10 0 0 0
 662
 663   where:
 664   1) 68 is the sendcount
 665   2) 68 10 10 10 is the recvcounts
 666   3) 0 is the root node
 667   4) 0 is the send datatype id, see decode_datatype()
 668   5) 0 is the recv datatype id, see decode_datatype()
 669   */
 670   double clock = smpi_process_simulated_elapsed();
 671   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 672   int send_size = parse_double(action[2]);
 673   int *disps = xbt_new0(int, comm_size);
 674   int *recvcounts = xbt_new0(int, comm_size);
 675   int i=0,recv_sum=0;
 676
 677   MPI_Datatype MPI_CURRENT_TYPE2;
 678   if(action[4+comm_size]) {
 679     MPI_CURRENT_TYPE=decode_datatype(action[4+comm_size]);
 680     MPI_CURRENT_TYPE2=decode_datatype(action[5+comm_size]);
 681   } else {
 682     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 683     MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
 684   }
 685   void *send = calloc(send_size, smpi_datatype_size(MPI_CURRENT_TYPE));
 686   void *recv = NULL;
 687   for(i=0;i<comm_size;i++) {
 688     recvcounts[i] = atoi(action[i+3]);
 689     recv_sum=recv_sum+recvcounts[i];
 690     disps[i] = 0;
 691   }
 692
 693   int root=atoi(action[3+comm_size]);
 694   int rank = smpi_process_index();
 695
 696   if(rank==root)
 697     recv = calloc(recv_sum, smpi_datatype_size(MPI_CURRENT_TYPE2));
 698
 699 #ifdef HAVE_TRACING
 700   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 701   extra->type = TRACING_GATHERV;
 702   extra->send_size = send_size;
 703   extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
 704   for(i=0; i< comm_size; i++)//copy data to avoid bad free
 705     extra->recvcounts[i] = recvcounts[i];
 706   extra->root = root;
 707   extra->num_processes = comm_size;
 708   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 709   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
 710
 711   TRACE_smpi_collective_in(rank, root, __FUNCTION__, extra);
 712 #endif
 713 smpi_mpi_gatherv(send, send_size, MPI_CURRENT_TYPE,
 714                 recv, recvcounts, disps, MPI_CURRENT_TYPE2,
 715                 root, MPI_COMM_WORLD);
 716
 717 #ifdef HAVE_TRACING
 718   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 719 #endif
 720
 721   log_timed_action (action, clock);
 722   xbt_free(recvcounts);
 723   xbt_free(send);
 724   if(recv)xbt_free(recv);
 725   xbt_free(disps);
 726
 727 }
 728
 729 static void action_reducescatter(const char *const *action) {
 730
 731     /*
 732  The structure of the reducescatter action for the rank 0 (total 4 processes)
 733  is the following:
 734 0 reduceScatter 275427 275427 275427 204020 11346849 0
 735
 736   where:
 737   1) The first four values after the name of the action declare the recvcounts array
 738   2) The value 11346849 is the amount of instructions
 739   3) The last value corresponds to the datatype, see decode_datatype().
 740
 741   We analyze a MPI_Reduce_scatter call to one MPI_Reduce and one MPI_Scatterv.
 742
 743    */
 744
 745   double clock = smpi_process_simulated_elapsed();
 746   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 747   int comp_size = parse_double(action[2+comm_size]);
 748   int *recvcounts = xbt_new0(int, comm_size);
 749   int *disps = xbt_new0(int, comm_size);
 750   int i=0,recv_sum=0;
 751   int root=0;
 752   int rank = smpi_process_index();
 753
 754   if(action[3+comm_size])
 755     MPI_CURRENT_TYPE=decode_datatype(action[3+comm_size]);
 756   else
 757     MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 758
 759   for(i=0;i<comm_size;i++) {
 760     recvcounts[i] = atoi(action[i+2]);
 761     recv_sum=recv_sum+recvcounts[i];
 762     disps[i] = 0;
 763   }
 764
 765 #ifdef HAVE_TRACING
 766   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 767   extra->type = TRACING_REDUCE_SCATTER;
 768   extra->send_size = 0;
 769   extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
 770   for(i=0; i< comm_size; i++)//copy data to avoid bad free
 771     extra->recvcounts[i] = recvcounts[i];
 772   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 773   extra->comp_size = comp_size;
 774   extra->num_processes = comm_size;
 775
 776
 777   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 778 #endif
 779    mpi_coll_reduce_fun(NULL, NULL, recv_sum, MPI_CURRENT_TYPE, MPI_OP_NULL,
 780        root, MPI_COMM_WORLD);
 781    smpi_mpi_scatterv(NULL, recvcounts, disps, MPI_CURRENT_TYPE, NULL,
 782                       recvcounts[rank], MPI_CURRENT_TYPE, 0, MPI_COMM_WORLD);
 783    smpi_execute_flops(comp_size);
 784
 785
 786 #ifdef HAVE_TRACING
 787   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 788 #endif
 789   xbt_free(recvcounts);
 790   xbt_free(disps);
 791   log_timed_action (action, clock);
 792 }
 793
 794
 795 static void action_allgatherv(const char *const *action) {
 796
 797   /*
 798  The structure of the allgatherv action for the rank 0 (total 4 processes)
 799  is the following:
 800 0 allGatherV 275427 275427 275427 275427 204020
 801
 802   where:
 803   1) 275427 is the sendcount
 804   2) The next four elements declare the recvcounts array
 805   3) No more values mean that the datatype for sent and receive buffer
 806   is the default one, see decode_datatype().
 807
 808    */
 809
 810   double clock = smpi_process_simulated_elapsed();
 811
 812   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 813   int i=0;
 814   int sendcount=atoi(action[2]);
 815   int *recvcounts = xbt_new0(int, comm_size);
 816   int *disps = xbt_new0(int, comm_size);
 817   int recv_sum=0;
 818   MPI_Datatype MPI_CURRENT_TYPE2;
 819
 820   if(action[3+comm_size]) {
 821     MPI_CURRENT_TYPE = decode_datatype(action[3+comm_size]);
 822     MPI_CURRENT_TYPE2 = decode_datatype(action[4+comm_size]);
 823   } else {
 824     MPI_CURRENT_TYPE = MPI_DEFAULT_TYPE;
 825     MPI_CURRENT_TYPE2 = MPI_DEFAULT_TYPE;
 826   }
 827   void *sendbuf = calloc(sendcount, smpi_datatype_size(MPI_CURRENT_TYPE));
 828
 829   for(i=0;i<comm_size;i++) {
 830     recvcounts[i] = atoi(action[i+3]);
 831     recv_sum=recv_sum+recvcounts[i];
 832   }
 833   void *recvbuf = calloc(recv_sum, smpi_datatype_size(MPI_CURRENT_TYPE2));
 834
 835 #ifdef HAVE_TRACING
 836   int rank = smpi_process_index();
 837   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 838   extra->type = TRACING_ALLGATHERV;
 839   extra->send_size = sendcount;
 840   extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
 841   for(i=0; i< comm_size; i++)//copy data to avoid bad free
 842     extra->recvcounts[i] = recvcounts[i];
 843   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 844   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
 845   extra->num_processes = comm_size;
 846
 847   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 848 #endif
 849
 850 mpi_coll_allgatherv_fun(sendbuf, sendcount, MPI_CURRENT_TYPE, recvbuf, recvcounts, disps, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
 851
 852 #ifdef HAVE_TRACING
 853   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 854 #endif
 855
 856   log_timed_action (action, clock);
 857   xbt_free(sendbuf);
 858   xbt_free(recvbuf);
 859   xbt_free(recvcounts);
 860   xbt_free(disps);
 861 }
 862
 863
 864 static void action_allToAllv(const char *const *action) {
 865   /*
 866  The structure of the allToAllV action for the rank 0 (total 4 processes)
 867  is the following:
 868   0 allToAllV 100 1 7 10 12 100 1 70 10 5
 869
 870   where:
 871   1) 100 is the size of the send buffer *sizeof(int),
 872   2) 1 7 10 12 is the sendcounts array
 873   3) 100*sizeof(int) is the size of the receiver buffer
 874   4)  1 70 10 5 is the recvcounts array
 875
 876    */
 877
 878
 879   double clock = smpi_process_simulated_elapsed();
 880
 881   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 882   int send_buf_size=0,recv_buf_size=0,i=0;
 883   int *sendcounts = xbt_new0(int, comm_size);
 884   int *recvcounts = xbt_new0(int, comm_size);
 885   int *senddisps = xbt_new0(int, comm_size);
 886   int *recvdisps = xbt_new0(int, comm_size);
 887
 888   MPI_Datatype MPI_CURRENT_TYPE2;
 889
 890   send_buf_size=parse_double(action[2]);
 891   recv_buf_size=parse_double(action[3+comm_size]);
 892   if(action[4+2*comm_size]) {
 893     MPI_CURRENT_TYPE=decode_datatype(action[4+2*comm_size]);
 894     MPI_CURRENT_TYPE2=decode_datatype(action[5+2*comm_size]);
 895   }
 896   else {
 897       MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 898       MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
 899   }
 900
 901   void *sendbuf = calloc(send_buf_size, smpi_datatype_size(MPI_CURRENT_TYPE));
 902   void *recvbuf = calloc(recv_buf_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
 903
 904   for(i=0;i<comm_size;i++) {
 905     sendcounts[i] = atoi(action[i+3]);
 906     recvcounts[i] = atoi(action[i+4+comm_size]);
 907   }
 908
 909
 910 #ifdef HAVE_TRACING
 911   int rank = smpi_process_index();
 912   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 913   extra->type = TRACING_ALLTOALLV;
 914   extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
 915   extra->sendcounts= xbt_malloc(comm_size*sizeof(int));
 916   extra->num_processes = comm_size;
 917
 918   for(i=0; i< comm_size; i++){//copy data to avoid bad free
 919     extra->send_size += sendcounts[i];
 920     extra->sendcounts[i] = sendcounts[i];
 921     extra->recv_size += recvcounts[i];
 922     extra->recvcounts[i] = recvcounts[i];
 923   }
 924   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 925   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
 926
 927   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 928 #endif
 929     mpi_coll_alltoallv_fun(sendbuf, sendcounts, senddisps, MPI_CURRENT_TYPE,
 930                                recvbuf, recvcounts, recvdisps, MPI_CURRENT_TYPE,
 931                                MPI_COMM_WORLD);
 932 #ifdef HAVE_TRACING
 933   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 934 #endif
 935
 936   log_timed_action (action, clock);
 937   xbt_free(sendbuf);
 938   xbt_free(recvbuf);
 939   xbt_free(sendcounts);
 940   xbt_free(recvcounts);
 941   xbt_free(senddisps);
 942   xbt_free(recvdisps);
 943 }
 944
 945 void smpi_replay_init(int *argc, char***argv){
 946   smpi_process_init(argc, argv);
 947   smpi_process_mark_as_initialized();
 948 #ifdef HAVE_TRACING
 949   int rank = smpi_process_index();
 950   TRACE_smpi_init(rank);
 951   TRACE_smpi_computing_init(rank);
 952   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 953   extra->type = TRACING_INIT;
 954   TRACE_smpi_collective_in(rank, -1, __FUNCTION__, extra);
 955   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 956 #endif
 957
 958   if (!smpi_process_index()){
 959     _xbt_replay_action_init();
 960     xbt_replay_action_register("init",       action_init);
 961     xbt_replay_action_register("finalize",   action_finalize);
 962     xbt_replay_action_register("comm_size",  action_comm_size);
 963     xbt_replay_action_register("comm_split", action_comm_split);
 964     xbt_replay_action_register("comm_dup",   action_comm_dup);
 965     xbt_replay_action_register("send",       action_send);
 966     xbt_replay_action_register("Isend",      action_Isend);
 967     xbt_replay_action_register("recv",       action_recv);
 968     xbt_replay_action_register("Irecv",      action_Irecv);
 969     xbt_replay_action_register("wait",       action_wait);
 970     xbt_replay_action_register("waitAll",    action_waitall);
 971     xbt_replay_action_register("barrier",    action_barrier);
 972     xbt_replay_action_register("bcast",      action_bcast);
 973     xbt_replay_action_register("reduce",     action_reduce);
 974     xbt_replay_action_register("allReduce",  action_allReduce);
 975     xbt_replay_action_register("allToAll",   action_allToAll);
 976     xbt_replay_action_register("allToAllV",  action_allToAllv);
 977     xbt_replay_action_register("gather",  action_gather);
 978     xbt_replay_action_register("gatherV",  action_gatherv);
 979     xbt_replay_action_register("allGatherV",  action_allgatherv);
 980     xbt_replay_action_register("reduceScatter",  action_reducescatter);
 981     xbt_replay_action_register("compute",    action_compute);
 982   }
 983
 984   xbt_replay_action_runner(*argc, *argv);
 985 }
 986
 987 int smpi_replay_finalize(){
 988   double sim_time= 1.;
 989   /* One active process will stop. Decrease the counter*/
 990   active_processes--;
 991   XBT_DEBUG("There are %lu elements in reqq[*]",
 992             xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]));
 993   xbt_dynar_free(&reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
 994   if(!active_processes){
 995     /* Last process alive speaking */
 996     /* end the simulated timer */
 997     sim_time = smpi_process_simulated_elapsed();
 998     XBT_INFO("Simulation time %g", sim_time);
 999     _xbt_replay_action_exit();
1000     xbt_free(reqq);
1001     reqq = NULL;
1002   }
1003   smpi_mpi_barrier(MPI_COMM_WORLD);
1004 #ifdef HAVE_TRACING
1005   int rank = smpi_process_index();
1006   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
1007   extra->type = TRACING_FINALIZE;
1008   TRACE_smpi_collective_in(rank, -1, __FUNCTION__, extra);
1009 #endif
1010   smpi_process_finalize();
1011 #ifdef HAVE_TRACING
1012   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
1013   TRACE_smpi_finalize(smpi_process_index());
1014 #endif
1015   smpi_process_destroy();
1016   return MPI_SUCCESS;
1017 }