Logo AND Algorithmique Numérique Distribuée

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