Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
[SMPI] (SECOND PATCH) Fix destinations/origins for tracing *TIData objects
[simgrid.git] / src / smpi / internals / smpi_replay.cpp
1 /* Copyright (c) 2009-2017. The SimGrid Team. All rights reserved.          */
2
3 /* This program is free software; you can redistribute it and/or modify it
4  * under the terms of the license (GNU LGPL) which comes with this package. */
5
6 #include "private.hpp"
7 #include "smpi_coll.hpp"
8 #include "smpi_comm.hpp"
9 #include "smpi_datatype.hpp"
10 #include "smpi_group.hpp"
11 #include "smpi_process.hpp"
12 #include "smpi_request.hpp"
13 #include "xbt/replay.hpp"
14
15 #include <unordered_map>
16 #include <vector>
17
18 #define KEY_SIZE (sizeof(int) * 2 + 1)
19
20 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_replay,smpi,"Trace Replay with SMPI");
21
22 int communicator_size = 0;
23 static int active_processes = 0;
24 std::unordered_map<int,std::vector<MPI_Request>*> reqq;
25
26 MPI_Datatype MPI_DEFAULT_TYPE;
27 MPI_Datatype MPI_CURRENT_TYPE;
28
29 static int sendbuffer_size=0;
30 char* sendbuffer=nullptr;
31 static int recvbuffer_size=0;
32 char* recvbuffer=nullptr;
33
34 static void log_timed_action (const char *const *action, double clock){
35   if (XBT_LOG_ISENABLED(smpi_replay, xbt_log_priority_verbose)){
36     char *name = xbt_str_join_array(action, " ");
37     XBT_VERB("%s %f", name, smpi_process()->simulated_elapsed()-clock);
38     xbt_free(name);
39   }
40 }
41
42 static std::vector<MPI_Request>* get_reqq_self()
43 {
44   return reqq.at(smpi_process()->index());
45 }
46
47 static void set_reqq_self(std::vector<MPI_Request> *mpi_request)
48 {
49    reqq.insert({smpi_process()->index(), mpi_request});
50 }
51
52 //allocate a single buffer for all sends, growing it if needed
53 void* smpi_get_tmp_sendbuffer(int size)
54 {
55   if (not smpi_process()->replaying())
56     return xbt_malloc(size);
57   if (sendbuffer_size<size){
58     sendbuffer=static_cast<char*>(xbt_realloc(sendbuffer,size));
59     sendbuffer_size=size;
60   }
61   return sendbuffer;
62 }
63
64 //allocate a single buffer for all recv
65 void* smpi_get_tmp_recvbuffer(int size){
66   if (not smpi_process()->replaying())
67     return xbt_malloc(size);
68   if (recvbuffer_size<size){
69     recvbuffer=static_cast<char*>(xbt_realloc(recvbuffer,size));
70     recvbuffer_size=size;
71   }
72   return recvbuffer;
73 }
74
75 void smpi_free_tmp_buffer(void* buf){
76   if (not smpi_process()->replaying())
77     xbt_free(buf);
78 }
79
80 /* Helper function */
81 static double parse_double(const char *string)
82 {
83   char *endptr;
84   double value = strtod(string, &endptr);
85   if (*endptr != '\0')
86     THROWF(unknown_error, 0, "%s is not a double", string);
87   return value;
88 }
89
90
91 //TODO: this logic should be moved inside the datatype class, to support all predefined types and get rid of is_replayable.
92 static MPI_Datatype decode_datatype(const char *const action)
93 {
94   switch(atoi(action)) {
95     case 0:
96       MPI_CURRENT_TYPE=MPI_DOUBLE;
97       break;
98     case 1:
99       MPI_CURRENT_TYPE=MPI_INT;
100       break;
101     case 2:
102       MPI_CURRENT_TYPE=MPI_CHAR;
103       break;
104     case 3:
105       MPI_CURRENT_TYPE=MPI_SHORT;
106       break;
107     case 4:
108       MPI_CURRENT_TYPE=MPI_LONG;
109       break;
110     case 5:
111       MPI_CURRENT_TYPE=MPI_FLOAT;
112       break;
113     case 6:
114       MPI_CURRENT_TYPE=MPI_BYTE;
115       break;
116     default:
117       MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
118       break;
119   }
120    return MPI_CURRENT_TYPE;
121 }
122
123 const char* encode_datatype(MPI_Datatype datatype)
124 {
125   if (datatype==MPI_BYTE)
126       return "";
127   if(datatype==MPI_DOUBLE)
128       return "0";
129   if(datatype==MPI_INT)
130       return "1";
131   if(datatype==MPI_CHAR)
132       return "2";
133   if(datatype==MPI_SHORT)
134       return "3";
135   if(datatype==MPI_LONG)
136     return "4";
137   if(datatype==MPI_FLOAT)
138       return "5";
139   // default - not implemented.
140   // do not warn here as we pass in this function even for other trace formats
141   return "-1";
142 }
143
144 #define CHECK_ACTION_PARAMS(action, mandatory, optional) {\
145     int i=0;\
146     while(action[i]!=nullptr)\
147      i++;\
148     if(i<mandatory+2)                                           \
149     THROWF(arg_error, 0, "%s replay failed.\n" \
150           "%d items were given on the line. First two should be process_id and action.  " \
151           "This action needs after them %d mandatory arguments, and accepts %d optional ones. \n" \
152           "Please contact the Simgrid team if support is needed", __FUNCTION__, i, mandatory, optional);\
153   }
154
155 namespace simgrid {
156 namespace smpi {
157
158 static void action_init(const char *const *action)
159 {
160   XBT_DEBUG("Initialize the counters");
161   CHECK_ACTION_PARAMS(action, 0, 1)
162   if(action[2])
163     MPI_DEFAULT_TYPE = MPI_DOUBLE; // default MPE datatype
164   else
165     MPI_DEFAULT_TYPE = MPI_BYTE; // default TAU datatype
166
167   /* start a simulated timer */
168   smpi_process()->simulated_start();
169   /*initialize the number of active processes */
170   active_processes = smpi_process_count();
171
172   set_reqq_self(new std::vector<MPI_Request>);
173 }
174
175 static void action_finalize(const char *const *action)
176 {
177   /* Nothing to do */
178 }
179
180 static void action_comm_size(const char *const *action)
181 {
182   communicator_size = parse_double(action[2]);
183   log_timed_action (action, smpi_process()->simulated_elapsed());
184 }
185
186 static void action_comm_split(const char *const *action)
187 {
188   log_timed_action (action, smpi_process()->simulated_elapsed());
189 }
190
191 static void action_comm_dup(const char *const *action)
192 {
193   log_timed_action (action, smpi_process()->simulated_elapsed());
194 }
195
196 static void action_compute(const char *const *action)
197 {
198   CHECK_ACTION_PARAMS(action, 1, 0)
199   double clock = smpi_process()->simulated_elapsed();
200   double flops= parse_double(action[2]);
201   int rank = smpi_process()->index();
202
203   TRACE_smpi_computing_in(rank, flops);
204   smpi_execute_flops(flops);
205   TRACE_smpi_computing_out(rank);
206
207   log_timed_action (action, clock);
208 }
209
210 static void action_send(const char *const *action)
211 {
212   CHECK_ACTION_PARAMS(action, 2, 1)
213   int to = atoi(action[2]);
214   double size=parse_double(action[3]);
215   double clock = smpi_process()->simulated_elapsed();
216
217   if(action[4])
218     MPI_CURRENT_TYPE=decode_datatype(action[4]);
219   else
220     MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
221
222   int rank = smpi_process()->index();
223   int dst_traced = MPI_COMM_WORLD->group()->rank(to);
224
225   TRACE_smpi_comm_in(rank, __FUNCTION__,
226                      new simgrid::instr::Pt2PtTIData("send", to, size, encode_datatype(MPI_CURRENT_TYPE)));
227   if (not TRACE_smpi_view_internals())
228     TRACE_smpi_send(rank, rank, dst_traced, 0, size*MPI_CURRENT_TYPE->size());
229
230   Request::send(nullptr, size, MPI_CURRENT_TYPE, to , 0, MPI_COMM_WORLD);
231
232   TRACE_smpi_comm_out(rank);
233
234   log_timed_action(action, clock);
235 }
236
237 static void action_Isend(const char *const *action)
238 {
239   CHECK_ACTION_PARAMS(action, 2, 1)
240   int to = atoi(action[2]);
241   double size=parse_double(action[3]);
242   double clock = smpi_process()->simulated_elapsed();
243
244   if(action[4])
245     MPI_CURRENT_TYPE=decode_datatype(action[4]);
246   else
247     MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
248
249   int rank = smpi_process()->index();
250   int dst_traced = MPI_COMM_WORLD->group()->rank(to);
251   TRACE_smpi_comm_in(rank, __FUNCTION__,
252                      new simgrid::instr::Pt2PtTIData("Isend", to, size, encode_datatype(MPI_CURRENT_TYPE)));
253   if (not TRACE_smpi_view_internals())
254     TRACE_smpi_send(rank, rank, dst_traced, 0, size*MPI_CURRENT_TYPE->size());
255
256   MPI_Request request = Request::isend(nullptr, size, MPI_CURRENT_TYPE, to, 0, MPI_COMM_WORLD);
257
258   TRACE_smpi_comm_out(rank);
259
260   get_reqq_self()->push_back(request);
261
262   log_timed_action (action, clock);
263 }
264
265 static void action_recv(const char *const *action) {
266   CHECK_ACTION_PARAMS(action, 2, 1)
267   int from = atoi(action[2]);
268   double size=parse_double(action[3]);
269   double clock = smpi_process()->simulated_elapsed();
270   MPI_Status status;
271
272   if(action[4])
273     MPI_CURRENT_TYPE=decode_datatype(action[4]);
274   else
275     MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
276
277   int rank = smpi_process()->index();
278   int src_traced = MPI_COMM_WORLD->group()->rank(from);
279
280   TRACE_smpi_comm_in(rank, __FUNCTION__,
281                      new simgrid::instr::Pt2PtTIData("recv", from, size, encode_datatype(MPI_CURRENT_TYPE)));
282
283   //unknown size from the receiver point of view
284   if (size <= 0.0) {
285     Request::probe(from, 0, MPI_COMM_WORLD, &status);
286     size=status.count;
287   }
288
289   Request::recv(nullptr, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD, &status);
290
291   TRACE_smpi_comm_out(rank);
292   if (not TRACE_smpi_view_internals()) {
293     TRACE_smpi_recv(src_traced, rank, 0);
294   }
295
296   log_timed_action (action, clock);
297 }
298
299 static void action_Irecv(const char *const *action)
300 {
301   CHECK_ACTION_PARAMS(action, 2, 1)
302   int from = atoi(action[2]);
303   double size=parse_double(action[3]);
304   double clock = smpi_process()->simulated_elapsed();
305
306   if(action[4])
307     MPI_CURRENT_TYPE=decode_datatype(action[4]);
308   else
309     MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
310
311   int rank = smpi_process()->index();
312   int src_traced = MPI_COMM_WORLD->group()->rank(from);
313   TRACE_smpi_comm_in(rank, __FUNCTION__,
314                      new simgrid::instr::Pt2PtTIData("Irecv", from, size, encode_datatype(MPI_CURRENT_TYPE)));
315   MPI_Status status;
316   //unknow size from the receiver pov
317   if (size <= 0.0) {
318     Request::probe(from, 0, MPI_COMM_WORLD, &status);
319     size = status.count;
320   }
321
322   MPI_Request request = Request::irecv(nullptr, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD);
323
324   TRACE_smpi_comm_out(rank);
325   get_reqq_self()->push_back(request);
326
327   log_timed_action (action, clock);
328 }
329
330 static void action_test(const char* const* action)
331 {
332   CHECK_ACTION_PARAMS(action, 0, 0)
333   double clock = smpi_process()->simulated_elapsed();
334   MPI_Status status;
335
336   MPI_Request request = get_reqq_self()->back();
337   get_reqq_self()->pop_back();
338   //if request is null here, this may mean that a previous test has succeeded
339   //Different times in traced application and replayed version may lead to this
340   //In this case, ignore the extra calls.
341   if(request!=nullptr){
342     int rank = smpi_process()->index();
343     TRACE_smpi_testing_in(rank);
344
345     int flag = Request::test(&request, &status);
346
347     XBT_DEBUG("MPI_Test result: %d", flag);
348     /* push back request in vector to be caught by a subsequent wait. if the test did succeed, the request is now nullptr.*/
349     get_reqq_self()->push_back(request);
350
351     TRACE_smpi_testing_out(rank);
352   }
353   log_timed_action (action, clock);
354 }
355
356 static void action_wait(const char *const *action){
357   CHECK_ACTION_PARAMS(action, 0, 0)
358   double clock = smpi_process()->simulated_elapsed();
359   MPI_Status status;
360
361   xbt_assert(get_reqq_self()->size(), "action wait not preceded by any irecv or isend: %s",
362       xbt_str_join_array(action," "));
363   MPI_Request request = get_reqq_self()->back();
364   get_reqq_self()->pop_back();
365
366   if (request==nullptr){
367     /* Assume that the trace is well formed, meaning the comm might have been caught by a MPI_test. Then just return.*/
368     return;
369   }
370
371   int rank = request->comm() != MPI_COMM_NULL ? request->comm()->rank() : -1;
372
373   MPI_Group group = request->comm()->group();
374   int src_traced = group->rank(request->src());
375   int dst_traced = group->rank(request->dst());
376   int is_wait_for_receive = (request->flags() & RECV);
377   TRACE_smpi_comm_in(rank, __FUNCTION__, new simgrid::instr::NoOpTIData("wait"));
378
379   Request::wait(&request, &status);
380
381   TRACE_smpi_comm_out(rank);
382   if (is_wait_for_receive)
383     TRACE_smpi_recv(src_traced, dst_traced, 0);
384   log_timed_action (action, clock);
385 }
386
387 static void action_waitall(const char *const *action){
388   CHECK_ACTION_PARAMS(action, 0, 0)
389   double clock = smpi_process()->simulated_elapsed();
390   const unsigned int count_requests = get_reqq_self()->size();
391
392   if (count_requests>0) {
393     MPI_Status status[count_requests];
394
395    int rank_traced = smpi_process()->index();
396    TRACE_smpi_comm_in(rank_traced, __FUNCTION__, new simgrid::instr::Pt2PtTIData("waitAll", -1, count_requests, ""));
397    int recvs_snd[count_requests];
398    int recvs_rcv[count_requests];
399    for (unsigned int i = 0; i < count_requests; i++) {
400      const auto& req = (*get_reqq_self())[i];
401      if (req && (req->flags () & RECV)){
402        recvs_snd[i]=req->src();
403        recvs_rcv[i]=req->dst();
404      }else
405        recvs_snd[i]=-100;
406    }
407    Request::waitall(count_requests, &(*get_reqq_self())[0], status);
408
409    for (unsigned i = 0; i < count_requests; i++) {
410      if (recvs_snd[i]!=-100)
411        TRACE_smpi_recv(recvs_snd[i], recvs_rcv[i],0);
412    }
413    TRACE_smpi_comm_out(rank_traced);
414   }
415   log_timed_action (action, clock);
416 }
417
418 static void action_barrier(const char *const *action){
419   double clock = smpi_process()->simulated_elapsed();
420   int rank = smpi_process()->index();
421   TRACE_smpi_comm_in(rank, __FUNCTION__, new simgrid::instr::NoOpTIData("barrier"));
422
423   Colls::barrier(MPI_COMM_WORLD);
424
425   TRACE_smpi_comm_out(rank);
426   log_timed_action (action, clock);
427 }
428
429 static void action_bcast(const char *const *action)
430 {
431   CHECK_ACTION_PARAMS(action, 1, 2)
432   double size = parse_double(action[2]);
433   double clock = smpi_process()->simulated_elapsed();
434   int root=0;
435   /* Initialize MPI_CURRENT_TYPE in order to decrease the number of the checks */
436   MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
437
438   if(action[3]) {
439     root= atoi(action[3]);
440     if(action[4])
441       MPI_CURRENT_TYPE=decode_datatype(action[4]);
442   }
443
444   int rank = smpi_process()->index();
445   TRACE_smpi_comm_in(rank, __FUNCTION__,
446                      new simgrid::instr::CollTIData("bcast", MPI_COMM_WORLD->group()->actor(root)->getPid(), -1.0, size, -1,
447                                                     encode_datatype(MPI_CURRENT_TYPE), ""));
448
449   void *sendbuf = smpi_get_tmp_sendbuffer(size* MPI_CURRENT_TYPE->size());
450
451   Colls::bcast(sendbuf, size, MPI_CURRENT_TYPE, root, MPI_COMM_WORLD);
452
453   TRACE_smpi_comm_out(rank);
454   log_timed_action (action, clock);
455 }
456
457 static void action_reduce(const char *const *action)
458 {
459   CHECK_ACTION_PARAMS(action, 2, 2)
460   double comm_size = parse_double(action[2]);
461   double comp_size = parse_double(action[3]);
462   double clock = smpi_process()->simulated_elapsed();
463   int root=0;
464   MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
465
466   if(action[4]) {
467     root= atoi(action[4]);
468     if(action[5])
469       MPI_CURRENT_TYPE=decode_datatype(action[5]);
470   }
471
472   int rank = smpi_process()->index();
473   TRACE_smpi_comm_in(rank, __FUNCTION__,
474                      new simgrid::instr::CollTIData("reduce", MPI_COMM_WORLD->group()->actor(root)->getPid(), comp_size,
475                                                     comm_size, -1, encode_datatype(MPI_CURRENT_TYPE), ""));
476
477   void *recvbuf = smpi_get_tmp_sendbuffer(comm_size* MPI_CURRENT_TYPE->size());
478   void *sendbuf = smpi_get_tmp_sendbuffer(comm_size* MPI_CURRENT_TYPE->size());
479   Colls::reduce(sendbuf, recvbuf, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, root, MPI_COMM_WORLD);
480   smpi_execute_flops(comp_size);
481
482   TRACE_smpi_comm_out(rank);
483   log_timed_action (action, clock);
484 }
485
486 static void action_allReduce(const char *const *action) {
487   CHECK_ACTION_PARAMS(action, 2, 1)
488   double comm_size = parse_double(action[2]);
489   double comp_size = parse_double(action[3]);
490
491   if(action[4])
492     MPI_CURRENT_TYPE=decode_datatype(action[4]);
493   else
494     MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
495
496   double clock = smpi_process()->simulated_elapsed();
497   int rank = smpi_process()->index();
498   TRACE_smpi_comm_in(rank, __FUNCTION__, new simgrid::instr::CollTIData("allReduce", -1, comp_size, comm_size, -1,
499                                                                         encode_datatype(MPI_CURRENT_TYPE), ""));
500
501   void *recvbuf = smpi_get_tmp_sendbuffer(comm_size* MPI_CURRENT_TYPE->size());
502   void *sendbuf = smpi_get_tmp_sendbuffer(comm_size* MPI_CURRENT_TYPE->size());
503   Colls::allreduce(sendbuf, recvbuf, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, MPI_COMM_WORLD);
504   smpi_execute_flops(comp_size);
505
506   TRACE_smpi_comm_out(rank);
507   log_timed_action (action, clock);
508 }
509
510 static void action_allToAll(const char *const *action) {
511   CHECK_ACTION_PARAMS(action, 2, 2) //two mandatory (send and recv volumes) and two optional (corresponding datatypes)
512   double clock = smpi_process()->simulated_elapsed();
513   int comm_size = MPI_COMM_WORLD->size();
514   int send_size = parse_double(action[2]);
515   int recv_size = parse_double(action[3]);
516   MPI_Datatype MPI_CURRENT_TYPE2 = MPI_DEFAULT_TYPE;
517
518   if(action[4] && action[5]) {
519     MPI_CURRENT_TYPE=decode_datatype(action[4]);
520     MPI_CURRENT_TYPE2=decode_datatype(action[5]);
521   }
522   else
523     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
524
525   void *send = smpi_get_tmp_sendbuffer(send_size*comm_size* MPI_CURRENT_TYPE->size());
526   void *recv = smpi_get_tmp_recvbuffer(recv_size*comm_size* MPI_CURRENT_TYPE2->size());
527
528   int rank = smpi_process()->index();
529   TRACE_smpi_comm_in(rank, __FUNCTION__, new simgrid::instr::CollTIData("allToAll", -1, -1.0, send_size, recv_size,
530                                                                         encode_datatype(MPI_CURRENT_TYPE),
531                                                                         encode_datatype(MPI_CURRENT_TYPE2)));
532
533   Colls::alltoall(send, send_size, MPI_CURRENT_TYPE, recv, recv_size, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
534
535   TRACE_smpi_comm_out(rank);
536   log_timed_action (action, clock);
537 }
538
539 static void action_gather(const char *const *action) {
540   /* The structure of the gather action for the rank 0 (total 4 processes) is the following:
541         0 gather 68 68 0 0 0
542       where:
543         1) 68 is the sendcounts
544         2) 68 is the recvcounts
545         3) 0 is the root node
546         4) 0 is the send datatype id, see decode_datatype()
547         5) 0 is the recv datatype id, see decode_datatype()
548   */
549   CHECK_ACTION_PARAMS(action, 2, 3)
550   double clock = smpi_process()->simulated_elapsed();
551   int comm_size = MPI_COMM_WORLD->size();
552   int send_size = parse_double(action[2]);
553   int recv_size = parse_double(action[3]);
554   MPI_Datatype MPI_CURRENT_TYPE2 = MPI_DEFAULT_TYPE;
555   if(action[4] && action[5]) {
556     MPI_CURRENT_TYPE=decode_datatype(action[5]);
557     MPI_CURRENT_TYPE2=decode_datatype(action[6]);
558   } else {
559     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
560   }
561   void *send = smpi_get_tmp_sendbuffer(send_size* MPI_CURRENT_TYPE->size());
562   void *recv = nullptr;
563   int root=0;
564   if(action[4])
565     root=atoi(action[4]);
566   int rank = MPI_COMM_WORLD->rank();
567
568   if(rank==root)
569     recv = smpi_get_tmp_recvbuffer(recv_size*comm_size* MPI_CURRENT_TYPE2->size());
570
571   TRACE_smpi_comm_in(rank, __FUNCTION__, new simgrid::instr::CollTIData("gather", root, -1.0, send_size, recv_size,
572                                                                         encode_datatype(MPI_CURRENT_TYPE),
573                                                                         encode_datatype(MPI_CURRENT_TYPE2)));
574
575   Colls::gather(send, send_size, MPI_CURRENT_TYPE, recv, recv_size, MPI_CURRENT_TYPE2, root, MPI_COMM_WORLD);
576
577   TRACE_smpi_comm_out(smpi_process()->index());
578   log_timed_action (action, clock);
579 }
580
581 static void action_scatter(const char* const* action)
582 {
583   /* The structure of the scatter action for the rank 0 (total 4 processes) is the following:
584         0 gather 68 68 0 0 0
585       where:
586         1) 68 is the sendcounts
587         2) 68 is the recvcounts
588         3) 0 is the root node
589         4) 0 is the send datatype id, see decode_datatype()
590         5) 0 is the recv datatype id, see decode_datatype()
591   */
592   CHECK_ACTION_PARAMS(action, 2, 3)
593   double clock                   = smpi_process()->simulated_elapsed();
594   int comm_size                  = MPI_COMM_WORLD->size();
595   int send_size                  = parse_double(action[2]);
596   int recv_size                  = parse_double(action[3]);
597   MPI_Datatype MPI_CURRENT_TYPE2 = MPI_DEFAULT_TYPE;
598   if (action[4] && action[5]) {
599     MPI_CURRENT_TYPE  = decode_datatype(action[5]);
600     MPI_CURRENT_TYPE2 = decode_datatype(action[6]);
601   } else {
602     MPI_CURRENT_TYPE = MPI_DEFAULT_TYPE;
603   }
604   void* send = smpi_get_tmp_sendbuffer(send_size * MPI_CURRENT_TYPE->size());
605   void* recv = nullptr;
606   int root   = 0;
607   if (action[4])
608     root   = atoi(action[4]);
609   int rank = MPI_COMM_WORLD->rank();
610
611   if (rank == root)
612     recv = smpi_get_tmp_recvbuffer(recv_size * comm_size * MPI_CURRENT_TYPE2->size());
613
614   TRACE_smpi_comm_in(rank, __FUNCTION__, new simgrid::instr::CollTIData("gather", root, -1.0, send_size, recv_size,
615                                                                         encode_datatype(MPI_CURRENT_TYPE),
616                                                                         encode_datatype(MPI_CURRENT_TYPE2)));
617
618   Colls::scatter(send, send_size, MPI_CURRENT_TYPE, recv, recv_size, MPI_CURRENT_TYPE2, root, MPI_COMM_WORLD);
619
620   TRACE_smpi_comm_out(smpi_process()->index());
621   log_timed_action(action, clock);
622 }
623
624 static void action_gatherv(const char *const *action) {
625   /* The structure of the gatherv action for the rank 0 (total 4 processes) is the following:
626        0 gather 68 68 10 10 10 0 0 0
627      where:
628        1) 68 is the sendcount
629        2) 68 10 10 10 is the recvcounts
630        3) 0 is the root node
631        4) 0 is the send datatype id, see decode_datatype()
632        5) 0 is the recv datatype id, see decode_datatype()
633   */
634   double clock = smpi_process()->simulated_elapsed();
635   int comm_size = MPI_COMM_WORLD->size();
636   CHECK_ACTION_PARAMS(action, comm_size+1, 2)
637   int send_size = parse_double(action[2]);
638   int disps[comm_size];
639   int recvcounts[comm_size];
640   int recv_sum=0;
641
642   MPI_Datatype MPI_CURRENT_TYPE2 = MPI_DEFAULT_TYPE;
643   if(action[4+comm_size] && action[5+comm_size]) {
644     MPI_CURRENT_TYPE=decode_datatype(action[4+comm_size]);
645     MPI_CURRENT_TYPE2=decode_datatype(action[5+comm_size]);
646   } else
647     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
648
649   void *send = smpi_get_tmp_sendbuffer(send_size* MPI_CURRENT_TYPE->size());
650   void *recv = nullptr;
651   for(int i=0;i<comm_size;i++) {
652     recvcounts[i] = atoi(action[i+3]);
653     recv_sum=recv_sum+recvcounts[i];
654     disps[i]=0;
655   }
656
657   int root=atoi(action[3+comm_size]);
658   int rank = MPI_COMM_WORLD->rank();
659
660   if(rank==root)
661     recv = smpi_get_tmp_recvbuffer(recv_sum* MPI_CURRENT_TYPE2->size());
662
663   std::vector<int>* trace_recvcounts = new std::vector<int>;
664   for (int i = 0; i < comm_size; i++) // copy data to avoid bad free
665     trace_recvcounts->push_back(recvcounts[i]);
666
667   TRACE_smpi_comm_in(rank, __FUNCTION__, new simgrid::instr::VarCollTIData(
668                                              "gatherV", root, send_size, nullptr, -1, trace_recvcounts,
669                                              encode_datatype(MPI_CURRENT_TYPE), encode_datatype(MPI_CURRENT_TYPE2)));
670
671   Colls::gatherv(send, send_size, MPI_CURRENT_TYPE, recv, recvcounts, disps, MPI_CURRENT_TYPE2, root, MPI_COMM_WORLD);
672
673   TRACE_smpi_comm_out(smpi_process()->index());
674   log_timed_action (action, clock);
675 }
676
677 static void action_scatterv(const char* const* action)
678 {
679   /* The structure of the scatterv action for the rank 0 (total 4 processes) is the following:
680        0 gather 68 10 10 10 68 0 0 0
681      where:
682        1) 68 10 10 10 is the sendcounts
683        2) 68 is the recvcount
684        3) 0 is the root node
685        4) 0 is the send datatype id, see decode_datatype()
686        5) 0 is the recv datatype id, see decode_datatype()
687   */
688   double clock  = smpi_process()->simulated_elapsed();
689   int comm_size = MPI_COMM_WORLD->size();
690   CHECK_ACTION_PARAMS(action, comm_size + 1, 2)
691   int recv_size = parse_double(action[2 + comm_size]);
692   int disps[comm_size];
693   int sendcounts[comm_size];
694   int send_sum = 0;
695
696   MPI_Datatype MPI_CURRENT_TYPE2 = MPI_DEFAULT_TYPE;
697   if (action[4 + comm_size] && action[5 + comm_size]) {
698     MPI_CURRENT_TYPE  = decode_datatype(action[4 + comm_size]);
699     MPI_CURRENT_TYPE2 = decode_datatype(action[5 + comm_size]);
700   } else
701     MPI_CURRENT_TYPE = MPI_DEFAULT_TYPE;
702
703   void* send = nullptr;
704   void* recv = smpi_get_tmp_recvbuffer(recv_size * MPI_CURRENT_TYPE->size());
705   for (int i = 0; i < comm_size; i++) {
706     sendcounts[i] = atoi(action[i + 2]);
707     send_sum += sendcounts[i];
708     disps[i] = 0;
709   }
710
711   int root = atoi(action[3 + comm_size]);
712   int rank = MPI_COMM_WORLD->rank();
713
714   if (rank == root)
715     send = smpi_get_tmp_sendbuffer(send_sum * MPI_CURRENT_TYPE2->size());
716
717   std::vector<int>* trace_sendcounts = new std::vector<int>;
718   for (int i = 0; i < comm_size; i++) // copy data to avoid bad free
719     trace_sendcounts->push_back(sendcounts[i]);
720
721   TRACE_smpi_comm_in(rank, __FUNCTION__, new simgrid::instr::VarCollTIData(
722                                              "gatherV", root, -1, trace_sendcounts, recv_size, nullptr,
723                                              encode_datatype(MPI_CURRENT_TYPE), encode_datatype(MPI_CURRENT_TYPE2)));
724
725   Colls::scatterv(send, sendcounts, disps, MPI_CURRENT_TYPE, recv, recv_size, MPI_CURRENT_TYPE2, root, MPI_COMM_WORLD);
726
727   TRACE_smpi_comm_out(smpi_process()->index());
728   log_timed_action(action, clock);
729 }
730
731 static void action_reducescatter(const char *const *action) {
732  /* The structure of the reducescatter action for the rank 0 (total 4 processes) is the following:
733       0 reduceScatter 275427 275427 275427 204020 11346849 0
734     where:
735       1) The first four values after the name of the action declare the recvcounts array
736       2) The value 11346849 is the amount of instructions
737       3) The last value corresponds to the datatype, see decode_datatype().
738 */
739   double clock = smpi_process()->simulated_elapsed();
740   int comm_size = MPI_COMM_WORLD->size();
741   CHECK_ACTION_PARAMS(action, comm_size+1, 1)
742   int comp_size = parse_double(action[2+comm_size]);
743   int recvcounts[comm_size];
744   int rank = smpi_process()->index();
745   int size = 0;
746   std::vector<int>* trace_recvcounts = new std::vector<int>;
747   if(action[3+comm_size])
748     MPI_CURRENT_TYPE=decode_datatype(action[3+comm_size]);
749   else
750     MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
751
752   for(int i=0;i<comm_size;i++) {
753     recvcounts[i] = atoi(action[i+2]);
754     trace_recvcounts->push_back(recvcounts[i]);
755     size+=recvcounts[i];
756   }
757
758   TRACE_smpi_comm_in(rank, __FUNCTION__,
759                      new simgrid::instr::VarCollTIData("reduceScatter", -1, 0, nullptr, -1, trace_recvcounts,
760                                                        std::to_string(comp_size), /* ugly hack to print comp_size */
761                                                        encode_datatype(MPI_CURRENT_TYPE)));
762
763   void *sendbuf = smpi_get_tmp_sendbuffer(size* MPI_CURRENT_TYPE->size());
764   void *recvbuf = smpi_get_tmp_recvbuffer(size* MPI_CURRENT_TYPE->size());
765
766   Colls::reduce_scatter(sendbuf, recvbuf, recvcounts, MPI_CURRENT_TYPE, MPI_OP_NULL, MPI_COMM_WORLD);
767   smpi_execute_flops(comp_size);
768
769   TRACE_smpi_comm_out(rank);
770   log_timed_action (action, clock);
771 }
772
773 static void action_allgather(const char *const *action) {
774   /* The structure of the allgather action for the rank 0 (total 4 processes) is the following:
775         0 allGather 275427 275427
776     where:
777         1) 275427 is the sendcount
778         2) 275427 is the recvcount
779         3) No more values mean that the datatype for sent and receive buffer is the default one, see decode_datatype().
780   */
781   double clock = smpi_process()->simulated_elapsed();
782
783   CHECK_ACTION_PARAMS(action, 2, 2)
784   int sendcount=atoi(action[2]);
785   int recvcount=atoi(action[3]);
786
787   MPI_Datatype MPI_CURRENT_TYPE2 = MPI_DEFAULT_TYPE;
788
789   if(action[4] && action[5]) {
790     MPI_CURRENT_TYPE = decode_datatype(action[4]);
791     MPI_CURRENT_TYPE2 = decode_datatype(action[5]);
792   } else
793     MPI_CURRENT_TYPE = MPI_DEFAULT_TYPE;
794
795   void *sendbuf = smpi_get_tmp_sendbuffer(sendcount* MPI_CURRENT_TYPE->size());
796   void *recvbuf = smpi_get_tmp_recvbuffer(recvcount* MPI_CURRENT_TYPE2->size());
797
798   int rank = smpi_process()->index();
799
800   TRACE_smpi_comm_in(rank, __FUNCTION__, new simgrid::instr::CollTIData("allGather", -1, -1.0, sendcount, recvcount,
801                                                                         encode_datatype(MPI_CURRENT_TYPE),
802                                                                         encode_datatype(MPI_CURRENT_TYPE2)));
803
804   Colls::allgather(sendbuf, sendcount, MPI_CURRENT_TYPE, recvbuf, recvcount, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
805
806   TRACE_smpi_comm_out(rank);
807   log_timed_action (action, clock);
808 }
809
810 static void action_allgatherv(const char *const *action) {
811   /* The structure of the allgatherv action for the rank 0 (total 4 processes) is the following:
812         0 allGatherV 275427 275427 275427 275427 204020
813      where:
814         1) 275427 is the sendcount
815         2) The next four elements declare the recvcounts array
816         3) No more values mean that the datatype for sent and receive buffer is the default one, see decode_datatype().
817   */
818   double clock = smpi_process()->simulated_elapsed();
819
820   int comm_size = MPI_COMM_WORLD->size();
821   CHECK_ACTION_PARAMS(action, comm_size+1, 2)
822   int sendcount=atoi(action[2]);
823   int recvcounts[comm_size];
824   int disps[comm_size];
825   int recv_sum=0;
826   MPI_Datatype MPI_CURRENT_TYPE2 = MPI_DEFAULT_TYPE;
827
828   if(action[3+comm_size] && action[4+comm_size]) {
829     MPI_CURRENT_TYPE = decode_datatype(action[3+comm_size]);
830     MPI_CURRENT_TYPE2 = decode_datatype(action[4+comm_size]);
831   } else
832     MPI_CURRENT_TYPE = MPI_DEFAULT_TYPE;
833
834   void *sendbuf = smpi_get_tmp_sendbuffer(sendcount* MPI_CURRENT_TYPE->size());
835
836   for(int i=0;i<comm_size;i++) {
837     recvcounts[i] = atoi(action[i+3]);
838     recv_sum=recv_sum+recvcounts[i];
839     disps[i] = 0;
840   }
841   void *recvbuf = smpi_get_tmp_recvbuffer(recv_sum* MPI_CURRENT_TYPE2->size());
842
843   int rank = smpi_process()->index();
844
845   std::vector<int>* trace_recvcounts = new std::vector<int>;
846   for (int i = 0; i < comm_size; i++) // copy data to avoid bad free
847     trace_recvcounts->push_back(recvcounts[i]);
848
849   TRACE_smpi_comm_in(rank, __FUNCTION__, new simgrid::instr::VarCollTIData(
850                                              "allGatherV", -1, sendcount, nullptr, -1, trace_recvcounts,
851                                              encode_datatype(MPI_CURRENT_TYPE), encode_datatype(MPI_CURRENT_TYPE2)));
852
853   Colls::allgatherv(sendbuf, sendcount, MPI_CURRENT_TYPE, recvbuf, recvcounts, disps, MPI_CURRENT_TYPE2,
854                           MPI_COMM_WORLD);
855
856   TRACE_smpi_comm_out(rank);
857   log_timed_action (action, clock);
858 }
859
860 static void action_allToAllv(const char *const *action) {
861   /* The structure of the allToAllV action for the rank 0 (total 4 processes) is the following:
862         0 allToAllV 100 1 7 10 12 100 1 70 10 5
863      where:
864         1) 100 is the size of the send buffer *sizeof(int),
865         2) 1 7 10 12 is the sendcounts array
866         3) 100*sizeof(int) is the size of the receiver buffer
867         4)  1 70 10 5 is the recvcounts array
868   */
869   double clock = smpi_process()->simulated_elapsed();
870
871   int comm_size = MPI_COMM_WORLD->size();
872   CHECK_ACTION_PARAMS(action, 2*comm_size+2, 2)
873   int send_size = 0;
874   int recv_size = 0;
875   int sendcounts[comm_size];
876   std::vector<int>* trace_sendcounts = new std::vector<int>;
877   int recvcounts[comm_size];
878   std::vector<int>* trace_recvcounts = new std::vector<int>;
879   int senddisps[comm_size];
880   int recvdisps[comm_size];
881
882   MPI_Datatype MPI_CURRENT_TYPE2 = MPI_DEFAULT_TYPE;
883
884   int send_buf_size=parse_double(action[2]);
885   int recv_buf_size=parse_double(action[3+comm_size]);
886   if(action[4+2*comm_size] && action[5+2*comm_size]) {
887     MPI_CURRENT_TYPE=decode_datatype(action[4+2*comm_size]);
888     MPI_CURRENT_TYPE2=decode_datatype(action[5+2*comm_size]);
889   }
890   else
891     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
892
893   int rank       = smpi_process()->index();
894   void *sendbuf = smpi_get_tmp_sendbuffer(send_buf_size* MPI_CURRENT_TYPE->size());
895   void *recvbuf  = smpi_get_tmp_recvbuffer(recv_buf_size* MPI_CURRENT_TYPE2->size());
896
897   for(int i=0;i<comm_size;i++) {
898     sendcounts[i] = atoi(action[i+3]);
899     trace_sendcounts->push_back(sendcounts[i]);
900     send_size += sendcounts[i];
901     recvcounts[i] = atoi(action[i+4+comm_size]);
902     trace_recvcounts->push_back(recvcounts[i]);
903     recv_size += recvcounts[i];
904     senddisps[i] = 0;
905     recvdisps[i] = 0;
906   }
907
908   TRACE_smpi_comm_in(rank, __FUNCTION__, new simgrid::instr::VarCollTIData(
909                                              "allToAllV", -1, send_size, trace_sendcounts, recv_size, trace_recvcounts,
910                                              encode_datatype(MPI_CURRENT_TYPE), encode_datatype(MPI_CURRENT_TYPE2)));
911
912   Colls::alltoallv(sendbuf, sendcounts, senddisps, MPI_CURRENT_TYPE,recvbuf, recvcounts, recvdisps,
913                          MPI_CURRENT_TYPE, MPI_COMM_WORLD);
914
915   TRACE_smpi_comm_out(rank);
916   log_timed_action (action, clock);
917 }
918
919 }} // namespace simgrid::smpi
920
921 /** @brief Only initialize the replay, don't do it for real */
922 void smpi_replay_init(int* argc, char*** argv)
923 {
924   simgrid::smpi::Process::init(argc, argv);
925   smpi_process()->mark_as_initialized();
926   smpi_process()->set_replaying(true);
927
928   int rank = smpi_process()->index();
929   TRACE_smpi_init(rank);
930   TRACE_smpi_computing_init(rank);
931   TRACE_smpi_comm_in(rank, "smpi_replay_run_init", new simgrid::instr::NoOpTIData("init"));
932   TRACE_smpi_comm_out(rank);
933   xbt_replay_action_register("init",       simgrid::smpi::action_init);
934   xbt_replay_action_register("finalize",   simgrid::smpi::action_finalize);
935   xbt_replay_action_register("comm_size",  simgrid::smpi::action_comm_size);
936   xbt_replay_action_register("comm_split", simgrid::smpi::action_comm_split);
937   xbt_replay_action_register("comm_dup",   simgrid::smpi::action_comm_dup);
938   xbt_replay_action_register("send",       simgrid::smpi::action_send);
939   xbt_replay_action_register("Isend",      simgrid::smpi::action_Isend);
940   xbt_replay_action_register("recv",       simgrid::smpi::action_recv);
941   xbt_replay_action_register("Irecv",      simgrid::smpi::action_Irecv);
942   xbt_replay_action_register("test",       simgrid::smpi::action_test);
943   xbt_replay_action_register("wait",       simgrid::smpi::action_wait);
944   xbt_replay_action_register("waitAll",    simgrid::smpi::action_waitall);
945   xbt_replay_action_register("barrier",    simgrid::smpi::action_barrier);
946   xbt_replay_action_register("bcast",      simgrid::smpi::action_bcast);
947   xbt_replay_action_register("reduce",     simgrid::smpi::action_reduce);
948   xbt_replay_action_register("allReduce",  simgrid::smpi::action_allReduce);
949   xbt_replay_action_register("allToAll",   simgrid::smpi::action_allToAll);
950   xbt_replay_action_register("allToAllV",  simgrid::smpi::action_allToAllv);
951   xbt_replay_action_register("gather",     simgrid::smpi::action_gather);
952   xbt_replay_action_register("scatter", simgrid::smpi::action_scatter);
953   xbt_replay_action_register("gatherV",    simgrid::smpi::action_gatherv);
954   xbt_replay_action_register("scatterV", simgrid::smpi::action_scatterv);
955   xbt_replay_action_register("allGather",  simgrid::smpi::action_allgather);
956   xbt_replay_action_register("allGatherV", simgrid::smpi::action_allgatherv);
957   xbt_replay_action_register("reduceScatter",  simgrid::smpi::action_reducescatter);
958   xbt_replay_action_register("compute",    simgrid::smpi::action_compute);
959
960   //if we have a delayed start, sleep here.
961   if(*argc>2){
962     double value = xbt_str_parse_double((*argv)[2], "%s is not a double");
963     XBT_VERB("Delayed start for instance - Sleeping for %f flops ",value );
964     smpi_execute_flops(value);
965   } else {
966     //UGLY: force a context switch to be sure that all MSG_processes begin initialization
967     XBT_DEBUG("Force context switch by smpi_execute_flops  - Sleeping for 0.0 flops ");
968     smpi_execute_flops(0.0);
969   }
970 }
971
972 /** @brief actually run the replay after initialization */
973 void smpi_replay_main(int* argc, char*** argv)
974 {
975   simgrid::xbt::replay_runner(*argc, *argv);
976
977   /* and now, finalize everything */
978   /* One active process will stop. Decrease the counter*/
979   XBT_DEBUG("There are %zu elements in reqq[*]", get_reqq_self()->size());
980   if (not get_reqq_self()->empty()) {
981     unsigned int count_requests=get_reqq_self()->size();
982     MPI_Request requests[count_requests];
983     MPI_Status status[count_requests];
984     unsigned int i=0;
985
986     for (auto const& req : *get_reqq_self()) {
987       requests[i] = req;
988       i++;
989     }
990     simgrid::smpi::Request::waitall(count_requests, requests, status);
991   }
992   delete get_reqq_self();
993   active_processes--;
994
995   if(active_processes==0){
996     /* Last process alive speaking: end the simulated timer */
997     XBT_INFO("Simulation time %f", smpi_process()->simulated_elapsed());
998     xbt_free(sendbuffer);
999     xbt_free(recvbuffer);
1000   }
1001
1002   TRACE_smpi_comm_in(smpi_process()->index(), "smpi_replay_run_finalize", new simgrid::instr::NoOpTIData("finalize"));
1003
1004   smpi_process()->finalize();
1005
1006   TRACE_smpi_comm_out(smpi_process()->index());
1007   TRACE_smpi_finalize(smpi_process()->index());
1008 }
1009
1010 /** @brief chain a replay initialization and a replay start */
1011 void smpi_replay_run(int* argc, char*** argv)
1012 {
1013   smpi_replay_init(argc, argv);
1014   smpi_replay_main(argc, argv);
1015 }