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   xbt_assert(request != NULL, "found null request in globals->irecv");
 338 #ifdef HAVE_TRACING
 339   int rank = request->comm != MPI_COMM_NULL
 340       ? smpi_comm_rank(request->comm)
 341       : -1;
 342
 343   MPI_Group group = smpi_comm_group(request->comm);
 344   int src_traced = smpi_group_rank(group, request->src);
 345   int dst_traced = smpi_group_rank(group, request->dst);
 346   int is_wait_for_receive = request->recv;
 347   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 348   extra->type = TRACING_WAIT;
 349   TRACE_smpi_ptp_in(rank, src_traced, dst_traced, __FUNCTION__, extra);
 350 #endif
 351   smpi_mpi_wait(&request, &status);
 352 #ifdef HAVE_TRACING
 353   TRACE_smpi_ptp_out(rank, src_traced, dst_traced, __FUNCTION__);
 354   if (is_wait_for_receive) {
 355     TRACE_smpi_recv(rank, src_traced, dst_traced);
 356   }
 357 #endif
 358
 359   log_timed_action (action, clock);
 360 }
 361
 362 static void action_waitall(const char *const *action){
 363   double clock = smpi_process_simulated_elapsed();
 364   int count_requests=0;
 365   unsigned int i=0;
 366
 367   count_requests=xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
 368
 369   if (count_requests>0) {
 370     MPI_Request requests[count_requests];
 371     MPI_Status status[count_requests];
 372
 373     /*  The reqq is an array of dynars. Its index corresponds to the rank.
 374      Thus each rank saves its own requests to the array request. */
 375     xbt_dynar_foreach(reqq[smpi_comm_rank(MPI_COMM_WORLD)],i,requests[i]);
 376
 377   #ifdef HAVE_TRACING
 378    //save information from requests
 379
 380    xbt_dynar_t srcs = xbt_dynar_new(sizeof(int), NULL);
 381    xbt_dynar_t dsts = xbt_dynar_new(sizeof(int), NULL);
 382    xbt_dynar_t recvs = xbt_dynar_new(sizeof(int), NULL);
 383    for (i = 0; i < count_requests; i++) {
 384     if(requests[i]){
 385       int *asrc = xbt_new(int, 1);
 386       int *adst = xbt_new(int, 1);
 387       int *arecv = xbt_new(int, 1);
 388       *asrc = requests[i]->src;
 389       *adst = requests[i]->dst;
 390       *arecv = requests[i]->recv;
 391       xbt_dynar_insert_at(srcs, i, asrc);
 392       xbt_dynar_insert_at(dsts, i, adst);
 393       xbt_dynar_insert_at(recvs, i, arecv);
 394       xbt_free(asrc);
 395       xbt_free(adst);
 396       xbt_free(arecv);
 397     }else {
 398       int *t = xbt_new(int, 1);
 399       xbt_dynar_insert_at(srcs, i, t);
 400       xbt_dynar_insert_at(dsts, i, t);
 401       xbt_dynar_insert_at(recvs, i, t);
 402       xbt_free(t);
 403     }
 404    }
 405    int rank_traced = smpi_process_index();
 406    instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 407    extra->type = TRACING_WAITALL;
 408    extra->send_size=count_requests;
 409    TRACE_smpi_ptp_in(rank_traced, -1, -1, __FUNCTION__,extra);
 410  #endif
 411
 412     smpi_mpi_waitall(count_requests, requests, status);
 413
 414   #ifdef HAVE_TRACING
 415    for (i = 0; i < count_requests; i++) {
 416     int src_traced, dst_traced, is_wait_for_receive;
 417     xbt_dynar_get_cpy(srcs, i, &src_traced);
 418     xbt_dynar_get_cpy(dsts, i, &dst_traced);
 419     xbt_dynar_get_cpy(recvs, i, &is_wait_for_receive);
 420     if (is_wait_for_receive) {
 421       TRACE_smpi_recv(rank_traced, src_traced, dst_traced);
 422     }
 423    }
 424    TRACE_smpi_ptp_out(rank_traced, -1, -1, __FUNCTION__);
 425    //clean-up of dynars
 426    xbt_dynar_free(&srcs);
 427    xbt_dynar_free(&dsts);
 428    xbt_dynar_free(&recvs);
 429   #endif
 430
 431    xbt_dynar_reset(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
 432   }
 433   log_timed_action (action, clock);
 434 }
 435
 436 static void action_barrier(const char *const *action){
 437   double clock = smpi_process_simulated_elapsed();
 438 #ifdef HAVE_TRACING
 439   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 440   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 441   extra->type = TRACING_BARRIER;
 442   TRACE_smpi_collective_in(rank, -1, __FUNCTION__, extra);
 443 #endif
 444   smpi_mpi_barrier(MPI_COMM_WORLD);
 445 #ifdef HAVE_TRACING
 446   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 447 #endif
 448
 449   log_timed_action (action, clock);
 450 }
 451
 452
 453 static void action_bcast(const char *const *action)
 454 {
 455   double size = parse_double(action[2]);
 456   double clock = smpi_process_simulated_elapsed();
 457   int root=0;
 458   /*
 459    * Initialize MPI_CURRENT_TYPE in order to decrease
 460    * the number of the checks
 461    * */
 462   MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 463
 464   if(action[3]) {
 465     root= atoi(action[3]);
 466     if(action[4]) {
 467       MPI_CURRENT_TYPE=decode_datatype(action[4]);
 468     }
 469   }
 470
 471 #ifdef HAVE_TRACING
 472   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 473   int root_traced = smpi_group_index(smpi_comm_group(MPI_COMM_WORLD), root);
 474
 475   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 476   extra->type = TRACING_BCAST;
 477   extra->send_size = size;
 478   extra->root = root_traced;
 479   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 480   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__, extra);
 481
 482 #endif
 483
 484   mpi_coll_bcast_fun(NULL, size, MPI_CURRENT_TYPE, root, MPI_COMM_WORLD);
 485 #ifdef HAVE_TRACING
 486   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
 487 #endif
 488
 489   log_timed_action (action, clock);
 490 }
 491
 492 static void action_reduce(const char *const *action)
 493 {
 494   double comm_size = parse_double(action[2]);
 495   double comp_size = parse_double(action[3]);
 496   double clock = smpi_process_simulated_elapsed();
 497   int root=0;
 498   MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 499
 500   if(action[4]) {
 501     root= atoi(action[4]);
 502     if(action[5]) {
 503       MPI_CURRENT_TYPE=decode_datatype(action[5]);
 504     }
 505   }
 506
 507 #ifdef HAVE_TRACING
 508   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 509   int root_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), root);
 510   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 511   extra->type = TRACING_REDUCE;
 512   extra->send_size = comm_size;
 513   extra->comp_size = comp_size;
 514   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 515   extra->root = root_traced;
 516
 517   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__,extra);
 518 #endif
 519    mpi_coll_reduce_fun(NULL, NULL, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, root, MPI_COMM_WORLD);
 520    smpi_execute_flops(comp_size);
 521 #ifdef HAVE_TRACING
 522   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
 523 #endif
 524
 525   log_timed_action (action, clock);
 526 }
 527
 528 static void action_allReduce(const char *const *action) {
 529   double comm_size = parse_double(action[2]);
 530   double comp_size = parse_double(action[3]);
 531
 532   if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
 533   else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 534
 535   double clock = smpi_process_simulated_elapsed();
 536 #ifdef HAVE_TRACING
 537   int rank = smpi_comm_rank(MPI_COMM_WORLD);
 538   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 539   extra->type = TRACING_ALLREDUCE;
 540   extra->send_size = comm_size;
 541   extra->comp_size = comp_size;
 542   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 543
 544   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 545 #endif
 546   mpi_coll_reduce_fun(NULL, NULL, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, 0, MPI_COMM_WORLD);
 547   smpi_execute_flops(comp_size);
 548   mpi_coll_bcast_fun(NULL, comm_size, MPI_CURRENT_TYPE, 0, MPI_COMM_WORLD);
 549 #ifdef HAVE_TRACING
 550   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 551 #endif
 552
 553   log_timed_action (action, clock);
 554 }
 555
 556 static void action_allToAll(const char *const *action) {
 557   double clock = smpi_process_simulated_elapsed();
 558   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 559   int send_size = parse_double(action[2]);
 560   int recv_size = parse_double(action[3]);
 561   MPI_Datatype MPI_CURRENT_TYPE2;
 562
 563   if(action[4]) {
 564     MPI_CURRENT_TYPE=decode_datatype(action[4]);
 565     MPI_CURRENT_TYPE2=decode_datatype(action[5]);
 566   }
 567   else {
 568     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 569     MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
 570   }
 571   void *send = calloc(send_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE));
 572   void *recv = calloc(recv_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
 573
 574 #ifdef HAVE_TRACING
 575   int rank = smpi_process_index();
 576   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 577   extra->type = TRACING_ALLTOALL;
 578   extra->send_size = send_size;
 579   extra->recv_size = recv_size;
 580   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 581   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
 582
 583   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 584 #endif
 585
 586   mpi_coll_alltoall_fun(send, send_size, MPI_CURRENT_TYPE, recv, recv_size, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
 587
 588 #ifdef HAVE_TRACING
 589   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 590 #endif
 591
 592   log_timed_action (action, clock);
 593   xbt_free(send);
 594   xbt_free(recv);
 595 }
 596
 597
 598 static void action_gather(const char *const *action) {
 599   /*
 600  The structure of the gather action for the rank 0 (total 4 processes)
 601  is the following:
 602  0 gather 68 68 0 0 0
 603
 604   where:
 605   1) 68 is the sendcounts
 606   2) 68 is the recvcounts
 607   3) 0 is the root node
 608   4) 0 is the send datatype id, see decode_datatype()
 609   5) 0 is the recv datatype id, see decode_datatype()
 610   */
 611   double clock = smpi_process_simulated_elapsed();
 612   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 613   int send_size = parse_double(action[2]);
 614   int recv_size = parse_double(action[3]);
 615   MPI_Datatype MPI_CURRENT_TYPE2;
 616   if(action[5]) {
 617     MPI_CURRENT_TYPE=decode_datatype(action[5]);
 618     MPI_CURRENT_TYPE2=decode_datatype(action[6]);
 619   } else {
 620     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 621     MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
 622   }
 623   void *send = calloc(send_size, smpi_datatype_size(MPI_CURRENT_TYPE));
 624   void *recv = NULL;
 625
 626   int root=atoi(action[4]);
 627   int rank = smpi_process_index();
 628
 629   if(rank==root)
 630     recv = calloc(recv_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
 631
 632 #ifdef HAVE_TRACING
 633   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 634   extra->type = TRACING_GATHER;
 635   extra->send_size = send_size;
 636   extra->recv_size = recv_size;
 637   extra->root = root;
 638   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 639   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
 640
 641   TRACE_smpi_collective_in(rank, root, __FUNCTION__, extra);
 642 #endif
 643 smpi_mpi_gather(send, send_size, MPI_CURRENT_TYPE,
 644                 recv, recv_size, MPI_CURRENT_TYPE2,
 645                 root, MPI_COMM_WORLD);
 646
 647 #ifdef HAVE_TRACING
 648   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 649 #endif
 650
 651   log_timed_action (action, clock);
 652   xbt_free(send);
 653   xbt_free(recv);
 654 }
 655
 656
 657
 658 static void action_gatherv(const char *const *action) {
 659   /*
 660  The structure of the gatherv action for the rank 0 (total 4 processes)
 661  is the following:
 662  0 gather 68 68 10 10 10 0 0 0
 663
 664   where:
 665   1) 68 is the sendcount
 666   2) 68 10 10 10 is the recvcounts
 667   3) 0 is the root node
 668   4) 0 is the send datatype id, see decode_datatype()
 669   5) 0 is the recv datatype id, see decode_datatype()
 670   */
 671   double clock = smpi_process_simulated_elapsed();
 672   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 673   int send_size = parse_double(action[2]);
 674   int *disps = xbt_new0(int, comm_size);
 675   int *recvcounts = xbt_new0(int, comm_size);
 676   int i=0,recv_sum=0;
 677
 678   MPI_Datatype MPI_CURRENT_TYPE2;
 679   if(action[4+comm_size]) {
 680     MPI_CURRENT_TYPE=decode_datatype(action[4+comm_size]);
 681     MPI_CURRENT_TYPE2=decode_datatype(action[5+comm_size]);
 682   } else {
 683     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 684     MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
 685   }
 686   void *send = calloc(send_size, smpi_datatype_size(MPI_CURRENT_TYPE));
 687   void *recv = NULL;
 688   for(i=0;i<comm_size;i++) {
 689     recvcounts[i] = atoi(action[i+3]);
 690     recv_sum=recv_sum+recvcounts[i];
 691     disps[i] = 0;
 692   }
 693
 694   int root=atoi(action[3+comm_size]);
 695   int rank = smpi_process_index();
 696
 697   if(rank==root)
 698     recv = calloc(recv_sum, smpi_datatype_size(MPI_CURRENT_TYPE2));
 699
 700 #ifdef HAVE_TRACING
 701   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 702   extra->type = TRACING_GATHERV;
 703   extra->send_size = send_size;
 704   extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
 705   for(i=0; i< comm_size; i++)//copy data to avoid bad free
 706     extra->recvcounts[i] = recvcounts[i];
 707   extra->root = root;
 708   extra->num_processes = comm_size;
 709   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 710   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
 711
 712   TRACE_smpi_collective_in(rank, root, __FUNCTION__, extra);
 713 #endif
 714 smpi_mpi_gatherv(send, send_size, MPI_CURRENT_TYPE,
 715                 recv, recvcounts, disps, MPI_CURRENT_TYPE2,
 716                 root, MPI_COMM_WORLD);
 717
 718 #ifdef HAVE_TRACING
 719   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 720 #endif
 721
 722   log_timed_action (action, clock);
 723   xbt_free(recvcounts);
 724   xbt_free(send);
 725   xbt_free(recv);
 726   xbt_free(disps);
 727
 728 }
 729
 730 static void action_reducescatter(const char *const *action) {
 731
 732     /*
 733  The structure of the reducescatter action for the rank 0 (total 4 processes)
 734  is the following:
 735 0 reduceScatter 275427 275427 275427 204020 11346849 0
 736
 737   where:
 738   1) The first four values after the name of the action declare the recvcounts array
 739   2) The value 11346849 is the amount of instructions
 740   3) The last value corresponds to the datatype, see decode_datatype().
 741
 742   We analyze a MPI_Reduce_scatter call to one MPI_Reduce and one MPI_Scatterv.
 743
 744    */
 745
 746   double clock = smpi_process_simulated_elapsed();
 747   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 748   int comp_size = parse_double(action[2+comm_size]);
 749   int *recvcounts = xbt_new0(int, comm_size);
 750   int *disps = xbt_new0(int, comm_size);
 751   int i=0,recv_sum=0;
 752   int root=0;
 753   int rank = smpi_process_index();
 754
 755   if(action[3+comm_size])
 756     MPI_CURRENT_TYPE=decode_datatype(action[3+comm_size]);
 757   else
 758     MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
 759
 760   for(i=0;i<comm_size;i++) {
 761     recvcounts[i] = atoi(action[i+2]);
 762     recv_sum=recv_sum+recvcounts[i];
 763     disps[i] = 0;
 764   }
 765
 766 #ifdef HAVE_TRACING
 767   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 768   extra->type = TRACING_REDUCE_SCATTER;
 769   extra->send_size = 0;
 770   extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
 771   for(i=0; i< comm_size; i++)//copy data to avoid bad free
 772     extra->recvcounts[i] = recvcounts[i];
 773   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 774   extra->comp_size = comp_size;
 775   extra->num_processes = comm_size;
 776
 777
 778   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 779 #endif
 780    mpi_coll_reduce_fun(NULL, NULL, recv_sum, MPI_CURRENT_TYPE, MPI_OP_NULL,
 781        root, MPI_COMM_WORLD);
 782    smpi_mpi_scatterv(NULL, recvcounts, disps, MPI_CURRENT_TYPE, NULL,
 783                       recvcounts[rank], MPI_CURRENT_TYPE, 0, MPI_COMM_WORLD);
 784    smpi_execute_flops(comp_size);
 785
 786
 787 #ifdef HAVE_TRACING
 788   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 789 #endif
 790   xbt_free(recvcounts);
 791   xbt_free(disps);
 792   log_timed_action (action, clock);
 793 }
 794
 795
 796 static void action_allgatherv(const char *const *action) {
 797
 798   /*
 799  The structure of the allgatherv action for the rank 0 (total 4 processes)
 800  is the following:
 801 0 allGatherV 275427 275427 275427 275427 204020
 802
 803   where:
 804   1) 275427 is the sendcount
 805   2) The next four elements declare the recvcounts array
 806   3) No more values mean that the datatype for sent and receive buffer
 807   is the default one, see decode_datatype().
 808
 809    */
 810
 811   double clock = smpi_process_simulated_elapsed();
 812
 813   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 814   int i=0;
 815   int sendcount=atoi(action[2]);
 816   int *recvcounts = xbt_new0(int, comm_size);
 817   int *disps = xbt_new0(int, comm_size);
 818   int recv_sum=0;
 819   MPI_Datatype MPI_CURRENT_TYPE2;
 820
 821   if(action[3+comm_size]) {
 822     MPI_CURRENT_TYPE = decode_datatype(action[3+comm_size]);
 823     MPI_CURRENT_TYPE2 = decode_datatype(action[4+comm_size]);
 824   } else {
 825     MPI_CURRENT_TYPE = MPI_DEFAULT_TYPE;
 826     MPI_CURRENT_TYPE2 = MPI_DEFAULT_TYPE;
 827   }
 828   void *sendbuf = calloc(sendcount, smpi_datatype_size(MPI_CURRENT_TYPE));
 829
 830   for(i=0;i<comm_size;i++) {
 831     recvcounts[i] = atoi(action[i+3]);
 832     recv_sum=recv_sum+recvcounts[i];
 833   }
 834   void *recvbuf = calloc(recv_sum, smpi_datatype_size(MPI_CURRENT_TYPE2));
 835
 836 #ifdef HAVE_TRACING
 837   int rank = smpi_process_index();
 838   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 839   extra->type = TRACING_ALLGATHERV;
 840   extra->send_size = sendcount;
 841   extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
 842   for(i=0; i< comm_size; i++)//copy data to avoid bad free
 843     extra->recvcounts[i] = recvcounts[i];
 844   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 845   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
 846   extra->num_processes = comm_size;
 847
 848   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 849 #endif
 850
 851 mpi_coll_allgatherv_fun(sendbuf, sendcount, MPI_CURRENT_TYPE, recvbuf, recvcounts, disps, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
 852
 853 #ifdef HAVE_TRACING
 854   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 855 #endif
 856
 857   log_timed_action (action, clock);
 858   xbt_free(sendbuf);
 859   xbt_free(recvbuf);
 860   xbt_free(recvcounts);
 861   xbt_free(disps);
 862 }
 863
 864
 865 static void action_allToAllv(const char *const *action) {
 866   /*
 867  The structure of the allToAllV action for the rank 0 (total 4 processes)
 868  is the following:
 869   0 allToAllV 100 1 7 10 12 100 1 70 10 5
 870
 871   where:
 872   1) 100 is the size of the send buffer *sizeof(int),
 873   2) 1 7 10 12 is the sendcounts array
 874   3) 100*sizeof(int) is the size of the receiver buffer
 875   4)  1 70 10 5 is the recvcounts array
 876
 877    */
 878
 879
 880   double clock = smpi_process_simulated_elapsed();
 881
 882   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
 883   int send_buf_size=0,recv_buf_size=0,i=0;
 884   int *sendcounts = xbt_new0(int, comm_size);
 885   int *recvcounts = xbt_new0(int, comm_size);
 886   int *senddisps = xbt_new0(int, comm_size);
 887   int *recvdisps = xbt_new0(int, comm_size);
 888
 889   MPI_Datatype MPI_CURRENT_TYPE2;
 890
 891   send_buf_size=parse_double(action[2]);
 892   recv_buf_size=parse_double(action[3+comm_size]);
 893   if(action[4+2*comm_size]) {
 894     MPI_CURRENT_TYPE=decode_datatype(action[4+2*comm_size]);
 895     MPI_CURRENT_TYPE2=decode_datatype(action[5+2*comm_size]);
 896   }
 897   else {
 898       MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 899       MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
 900   }
 901
 902   void *sendbuf = calloc(send_buf_size, smpi_datatype_size(MPI_CURRENT_TYPE));
 903   void *recvbuf = calloc(recv_buf_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
 904
 905   for(i=0;i<comm_size;i++) {
 906     sendcounts[i] = atoi(action[i+3]);
 907     recvcounts[i] = atoi(action[i+4+comm_size]);
 908   }
 909
 910
 911 #ifdef HAVE_TRACING
 912   int rank = smpi_process_index();
 913   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 914   extra->type = TRACING_ALLTOALLV;
 915   extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
 916   extra->sendcounts= xbt_malloc(comm_size*sizeof(int));
 917   extra->num_processes = comm_size;
 918
 919   for(i=0; i< comm_size; i++){//copy data to avoid bad free
 920     extra->send_size += sendcounts[i];
 921     extra->sendcounts[i] = sendcounts[i];
 922     extra->recv_size += recvcounts[i];
 923     extra->recvcounts[i] = recvcounts[i];
 924   }
 925   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
 926   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
 927
 928   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 929 #endif
 930     mpi_coll_alltoallv_fun(sendbuf, sendcounts, senddisps, MPI_CURRENT_TYPE,
 931                                recvbuf, recvcounts, recvdisps, MPI_CURRENT_TYPE,
 932                                MPI_COMM_WORLD);
 933 #ifdef HAVE_TRACING
 934   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 935 #endif
 936
 937   log_timed_action (action, clock);
 938   xbt_free(sendbuf);
 939   xbt_free(recvbuf);
 940   xbt_free(sendcounts);
 941   xbt_free(recvcounts);
 942   xbt_free(senddisps);
 943   xbt_free(recvdisps);
 944 }
 945
 946 void smpi_replay_init(int *argc, char***argv){
 947   smpi_process_init(argc, argv);
 948   smpi_process_mark_as_initialized();
 949 #ifdef HAVE_TRACING
 950   int rank = smpi_process_index();
 951   TRACE_smpi_init(rank);
 952   TRACE_smpi_computing_init(rank);
 953   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
 954   extra->type = TRACING_INIT;
 955   TRACE_smpi_collective_in(rank, -1, __FUNCTION__, extra);
 956   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
 957 #endif
 958
 959   if (!smpi_process_index()){
 960     _xbt_replay_action_init();
 961     xbt_replay_action_register("init",       action_init);
 962     xbt_replay_action_register("finalize",   action_finalize);
 963     xbt_replay_action_register("comm_size",  action_comm_size);
 964     xbt_replay_action_register("comm_split", action_comm_split);
 965     xbt_replay_action_register("comm_dup",   action_comm_dup);
 966     xbt_replay_action_register("send",       action_send);
 967     xbt_replay_action_register("Isend",      action_Isend);
 968     xbt_replay_action_register("recv",       action_recv);
 969     xbt_replay_action_register("Irecv",      action_Irecv);
 970     xbt_replay_action_register("wait",       action_wait);
 971     xbt_replay_action_register("waitAll",    action_waitall);
 972     xbt_replay_action_register("barrier",    action_barrier);
 973     xbt_replay_action_register("bcast",      action_bcast);
 974     xbt_replay_action_register("reduce",     action_reduce);
 975     xbt_replay_action_register("allReduce",  action_allReduce);
 976     xbt_replay_action_register("allToAll",   action_allToAll);
 977     xbt_replay_action_register("allToAllV",  action_allToAllv);
 978     xbt_replay_action_register("gather",  action_gather);
 979     xbt_replay_action_register("gatherV",  action_gatherv);
 980     xbt_replay_action_register("allGatherV",  action_allgatherv);
 981     xbt_replay_action_register("reduceScatter",  action_reducescatter);
 982     xbt_replay_action_register("compute",    action_compute);
 983   }
 984
 985   xbt_replay_action_runner(*argc, *argv);
 986 }
 987
 988 int smpi_replay_finalize(){
 989   double sim_time= 1.;
 990   /* One active process will stop. Decrease the counter*/
 991   active_processes--;
 992   XBT_DEBUG("There are %lu elements in reqq[*]",
 993             xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]));
 994   xbt_dynar_free(&reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
 995   if(!active_processes){
 996     /* Last process alive speaking */
 997     /* end the simulated timer */
 998     sim_time = smpi_process_simulated_elapsed();
 999     XBT_INFO("Simulation time %g", sim_time);
1000     _xbt_replay_action_exit();
1001     xbt_free(reqq);
1002     reqq = NULL;
1003   }
1004   smpi_mpi_barrier(MPI_COMM_WORLD);
1005 #ifdef HAVE_TRACING
1006   int rank = smpi_process_index();
1007   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
1008   extra->type = TRACING_FINALIZE;
1009   TRACE_smpi_collective_in(rank, -1, __FUNCTION__, extra);
1010 #endif
1011   smpi_process_finalize();
1012 #ifdef HAVE_TRACING
1013   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
1014   TRACE_smpi_finalize(smpi_process_index());
1015 #endif
1016   smpi_process_destroy();
1017   return MPI_SUCCESS;
1018 }