Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
[SMPI] Replay: Fix datatype initialization
[simgrid.git] / src / smpi / internals / smpi_replay.cpp
1 /* Copyright (c) 2009-2018. 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 <boost/algorithm/string/join.hpp>
16 #include <memory>
17 #include <numeric>
18 #include <unordered_map>
19 #include <vector>
20
21 using simgrid::s4u::Actor;
22
23 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_replay,smpi,"Trace Replay with SMPI");
24
25 static int active_processes  = 0;
26 static std::unordered_map<int, std::vector<MPI_Request>*> reqq;
27
28 static MPI_Datatype MPI_DEFAULT_TYPE;
29
30 #define CHECK_ACTION_PARAMS(action, mandatory, optional)                                                               \
31   {                                                                                                                    \
32     if (action.size() < static_cast<unsigned long>(mandatory + 2))                                                     \
33       THROWF(arg_error, 0, "%s replay failed.\n"                                                                       \
34                            "%zu items were given on the line. First two should be process_id and action.  "            \
35                            "This action needs after them %lu mandatory arguments, and accepts %lu optional ones. \n"   \
36                            "Please contact the Simgrid team if support is needed",                                     \
37              __func__, action.size(), static_cast<unsigned long>(mandatory), static_cast<unsigned long>(optional));    \
38   }
39
40 static void log_timed_action(simgrid::xbt::ReplayAction& action, double clock)
41 {
42   if (XBT_LOG_ISENABLED(smpi_replay, xbt_log_priority_verbose)){
43     std::string s = boost::algorithm::join(action, " ");
44     XBT_VERB("%s %f", s.c_str(), smpi_process()->simulated_elapsed() - clock);
45   }
46 }
47
48 static std::vector<MPI_Request>* get_reqq_self()
49 {
50   return reqq.at(Actor::self()->getPid());
51 }
52
53 static void set_reqq_self(std::vector<MPI_Request> *mpi_request)
54 {
55    reqq.insert({Actor::self()->getPid(), mpi_request});
56 }
57
58 /* Helper function */
59 static double parse_double(std::string string)
60 {
61   return xbt_str_parse_double(string.c_str(), "%s is not a double");
62 }
63
64 namespace simgrid {
65 namespace smpi {
66
67 namespace Replay {
68 class ActionArgParser {
69 public:
70   virtual void parse(simgrid::xbt::ReplayAction& action, std::string name) { CHECK_ACTION_PARAMS(action, 0, 0) }
71 };
72
73 class SendRecvParser : public ActionArgParser {
74 public:
75   /* communication partner; if we send, this is the receiver and vice versa */
76   int partner;
77   double size;
78   MPI_Datatype datatype1 = MPI_DEFAULT_TYPE;
79
80   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
81   {
82     CHECK_ACTION_PARAMS(action, 2, 1)
83     partner = std::stoi(action[2]);
84     size    = parse_double(action[3]);
85     if (action.size() > 4)
86       datatype1 = simgrid::smpi::Datatype::decode(action[4]);
87   }
88 };
89
90 class ComputeParser : public ActionArgParser {
91 public:
92   /* communication partner; if we send, this is the receiver and vice versa */
93   double flops;
94
95   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
96   {
97     CHECK_ACTION_PARAMS(action, 1, 0)
98     flops = parse_double(action[2]);
99   }
100 };
101
102 class CollCommParser : public ActionArgParser {
103 public:
104   double size;
105   double comm_size;
106   double comp_size;
107   int send_size;
108   int recv_size;
109   int root = 0;
110   MPI_Datatype datatype1 = MPI_DEFAULT_TYPE;
111   MPI_Datatype datatype2 = MPI_DEFAULT_TYPE;
112 };
113
114 class BcastArgParser : public CollCommParser {
115 public:
116   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
117   {
118     CHECK_ACTION_PARAMS(action, 1, 2)
119     size = parse_double(action[2]);
120     root = (action.size() > 3) ? std::stoi(action[3]) : 0;
121     if (action.size() > 4)
122       datatype1 = simgrid::smpi::Datatype::decode(action[4]);
123   }
124 };
125
126 class ReduceArgParser : public CollCommParser {
127 public:
128   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
129   {
130     CHECK_ACTION_PARAMS(action, 2, 2)
131     comm_size = parse_double(action[2]);
132     comp_size = parse_double(action[3]);
133     root      = (action.size() > 4) ? std::stoi(action[4]) : 0;
134     if (action.size() > 5)
135       datatype1 = simgrid::smpi::Datatype::decode(action[5]);
136   }
137 };
138
139 class AllReduceArgParser : public CollCommParser {
140 public:
141   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
142   {
143     CHECK_ACTION_PARAMS(action, 2, 1)
144     comm_size = parse_double(action[2]);
145     comp_size = parse_double(action[3]);
146     if (action.size() > 4)
147       datatype1 = simgrid::smpi::Datatype::decode(action[4]);
148   }
149 };
150
151 class AllToAllArgParser : public CollCommParser {
152 public:
153   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
154   {
155     CHECK_ACTION_PARAMS(action, 2, 1)
156     comm_size = MPI_COMM_WORLD->size();
157     send_size = parse_double(action[2]);
158     recv_size = parse_double(action[3]);
159
160     if (action.size() > 4)
161       datatype1 = simgrid::smpi::Datatype::decode(action[4]);
162     if (action.size() > 5)
163       datatype2 = simgrid::smpi::Datatype::decode(action[5]);
164   }
165 };
166
167 class GatherArgParser : public CollCommParser {
168 public:
169   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
170   {
171     /* The structure of the gather action for the rank 0 (total 4 processes) is the following:
172           0 gather 68 68 0 0 0
173         where:
174           1) 68 is the sendcounts
175           2) 68 is the recvcounts
176           3) 0 is the root node
177           4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
178           5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
179     */
180     CHECK_ACTION_PARAMS(action, 2, 3)
181     comm_size = MPI_COMM_WORLD->size();
182     send_size = parse_double(action[2]);
183     recv_size = parse_double(action[3]);
184
185     if (name == "gather") {
186       root      = (action.size() > 4) ? std::stoi(action[4]) : 0;
187       if (action.size() > 5)
188         datatype1 = simgrid::smpi::Datatype::decode(action[5]);
189       if (action.size() > 6)
190         datatype2 = simgrid::smpi::Datatype::decode(action[6]);
191     }
192     else {
193       if (action.size() > 4)
194         datatype1 = simgrid::smpi::Datatype::decode(action[4]);
195       if (action.size() > 5)
196         datatype2 = simgrid::smpi::Datatype::decode(action[5]);
197     }
198   }
199 };
200
201 class GatherVArgParser : public CollCommParser {
202 public:
203   int recv_size_sum;
204   std::shared_ptr<std::vector<int>> recvcounts;
205   std::vector<int> disps;
206   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
207   {
208     /* The structure of the gatherv action for the rank 0 (total 4 processes) is the following:
209          0 gather 68 68 10 10 10 0 0 0
210        where:
211          1) 68 is the sendcount
212          2) 68 10 10 10 is the recvcounts
213          3) 0 is the root node
214          4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
215          5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
216     */
217     comm_size = MPI_COMM_WORLD->size();
218     CHECK_ACTION_PARAMS(action, comm_size+1, 2)
219     send_size = parse_double(action[2]);
220     disps     = std::vector<int>(comm_size, 0);
221     recvcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
222
223     if (name == "gatherV") {
224       root = (action.size() > 3 + comm_size) ? std::stoi(action[3 + comm_size]) : 0;
225       if (action.size() > 4 + comm_size)
226         datatype1 = simgrid::smpi::Datatype::decode(action[4 + comm_size]);
227       if (action.size() > 5 + comm_size)
228         datatype2 = simgrid::smpi::Datatype::decode(action[5 + comm_size]);
229     }
230     else {
231       int datatype_index = 0, disp_index = 0;
232       /* The 3 comes from "0 gather <sendcount>", which must always be present.
233        * The + comm_size is the recvcounts array, which must also be present
234        */
235       if (action.size() > 3 + comm_size + comm_size) { /* datatype + disp are specified */
236         datatype_index = 3 + comm_size;
237         disp_index     = datatype_index + 1;
238         datatype1      = simgrid::smpi::Datatype::decode(action[datatype_index]);
239         datatype2      = simgrid::smpi::Datatype::decode(action[datatype_index]);
240       } else if (action.size() > 3 + comm_size + 2) { /* disps specified; datatype is not specified; use the default one */
241         disp_index     = 3 + comm_size;
242       } else if (action.size() > 3 + comm_size)  { /* only datatype, no disp specified */
243         datatype_index = 3 + comm_size;
244         datatype1      = simgrid::smpi::Datatype::decode(action[datatype_index]);
245         datatype2      = simgrid::smpi::Datatype::decode(action[datatype_index]);
246       }
247
248       if (disp_index != 0) {
249         for (unsigned int i = 0; i < comm_size; i++)
250           disps[i]          = std::stoi(action[disp_index + i]);
251       }
252     }
253
254     for (unsigned int i = 0; i < comm_size; i++) {
255       (*recvcounts)[i] = std::stoi(action[i + 3]);
256     }
257     recv_size_sum = std::accumulate(recvcounts->begin(), recvcounts->end(), 0);
258   }
259 };
260
261 class ScatterArgParser : public CollCommParser {
262 public:
263   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
264   {
265     /* The structure of the scatter action for the rank 0 (total 4 processes) is the following:
266           0 gather 68 68 0 0 0
267         where:
268           1) 68 is the sendcounts
269           2) 68 is the recvcounts
270           3) 0 is the root node
271           4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
272           5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
273     */
274     CHECK_ACTION_PARAMS(action, 2, 3)
275     comm_size   = MPI_COMM_WORLD->size();
276     send_size   = parse_double(action[2]);
277     recv_size   = parse_double(action[3]);
278     root   = (action.size() > 4) ? std::stoi(action[4]) : 0;
279     if (action.size() > 5)
280       datatype1 = simgrid::smpi::Datatype::decode(action[5]);
281     if (action.size() > 6)
282       datatype2 = simgrid::smpi::Datatype::decode(action[6]);
283   }
284 };
285
286 class ScatterVArgParser : public CollCommParser {
287 public:
288   int recv_size_sum;
289   int send_size_sum;
290   std::shared_ptr<std::vector<int>> sendcounts;
291   std::vector<int> disps;
292   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
293   {
294     /* The structure of the scatterv action for the rank 0 (total 4 processes) is the following:
295        0 gather 68 10 10 10 68 0 0 0
296         where:
297         1) 68 10 10 10 is the sendcounts
298         2) 68 is the recvcount
299         3) 0 is the root node
300         4) 0 is the send datatype id, see simgrid::smpi::Datatype::decode()
301         5) 0 is the recv datatype id, see simgrid::smpi::Datatype::decode()
302     */
303     CHECK_ACTION_PARAMS(action, comm_size + 1, 2)
304     recv_size  = parse_double(action[2 + comm_size]);
305     disps      = std::vector<int>(comm_size, 0);
306     sendcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
307
308     if (action.size() > 5 + comm_size)
309       datatype1 = simgrid::smpi::Datatype::decode(action[4 + comm_size]);
310     if (action.size() > 5 + comm_size)
311       datatype2 = simgrid::smpi::Datatype::decode(action[5]);
312
313     for (unsigned int i = 0; i < comm_size; i++) {
314       (*sendcounts)[i] = std::stoi(action[i + 2]);
315     }
316     send_size_sum = std::accumulate(sendcounts->begin(), sendcounts->end(), 0);
317     root = (action.size() > 3 + comm_size) ? std::stoi(action[3 + comm_size]) : 0;
318   }
319 };
320
321 class ReduceScatterArgParser : public CollCommParser {
322 public:
323   int recv_size_sum;
324   std::shared_ptr<std::vector<int>> recvcounts;
325   std::vector<int> disps;
326   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
327   {
328     /* The structure of the reducescatter action for the rank 0 (total 4 processes) is the following:
329          0 reduceScatter 275427 275427 275427 204020 11346849 0
330        where:
331          1) The first four values after the name of the action declare the recvcounts array
332          2) The value 11346849 is the amount of instructions
333          3) The last value corresponds to the datatype, see simgrid::smpi::Datatype::decode().
334     */
335     comm_size = MPI_COMM_WORLD->size();
336     CHECK_ACTION_PARAMS(action, comm_size+1, 1)
337     comp_size = parse_double(action[2+comm_size]);
338     recvcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
339     if (action.size() > 3 + comm_size)
340       datatype1 = simgrid::smpi::Datatype::decode(action[3 + comm_size]);
341
342     for (unsigned int i = 0; i < comm_size; i++) {
343       recvcounts->push_back(std::stoi(action[i + 2]));
344     }
345     recv_size_sum = std::accumulate(recvcounts->begin(), recvcounts->end(), 0);
346   }
347 };
348
349 class AllToAllVArgParser : public CollCommParser {
350 public:
351   int recv_size_sum;
352   int send_size_sum;
353   std::shared_ptr<std::vector<int>> recvcounts;
354   std::shared_ptr<std::vector<int>> sendcounts;
355   std::vector<int> senddisps;
356   std::vector<int> recvdisps;
357   int send_buf_size;
358   int recv_buf_size;
359   void parse(simgrid::xbt::ReplayAction& action, std::string name) override
360   {
361     /* The structure of the allToAllV action for the rank 0 (total 4 processes) is the following:
362           0 allToAllV 100 1 7 10 12 100 1 70 10 5
363        where:
364         1) 100 is the size of the send buffer *sizeof(int),
365         2) 1 7 10 12 is the sendcounts array
366         3) 100*sizeof(int) is the size of the receiver buffer
367         4)  1 70 10 5 is the recvcounts array
368     */
369     comm_size = MPI_COMM_WORLD->size();
370     CHECK_ACTION_PARAMS(action, 2*comm_size+2, 2)
371     sendcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
372     recvcounts = std::shared_ptr<std::vector<int>>(new std::vector<int>(comm_size));
373     senddisps  = std::vector<int>(comm_size, 0);
374     recvdisps  = std::vector<int>(comm_size, 0);
375
376     if (action.size() > 5 + 2 * comm_size)
377       datatype1 = simgrid::smpi::Datatype::decode(action[4 + 2 * comm_size]);
378     if (action.size() > 5 + 2 * comm_size)
379       datatype2 = simgrid::smpi::Datatype::decode(action[5 + 2 * comm_size]);
380
381     send_buf_size=parse_double(action[2]);
382     recv_buf_size=parse_double(action[3+comm_size]);
383     for (unsigned int i = 0; i < comm_size; i++) {
384       (*sendcounts)[i] = std::stoi(action[3 + i]);
385       (*recvcounts)[i] = std::stoi(action[4 + comm_size + i]);
386     }
387     send_size_sum = std::accumulate(sendcounts->begin(), sendcounts->end(), 0);
388     recv_size_sum = std::accumulate(recvcounts->begin(), recvcounts->end(), 0);
389   }
390 };
391
392 template <class T> class ReplayAction {
393 protected:
394   const std::string name;
395   T args;
396
397   int my_proc_id;
398
399 public:
400   explicit ReplayAction(std::string name) : name(name), my_proc_id(simgrid::s4u::Actor::self()->getPid()) {}
401
402   virtual void execute(simgrid::xbt::ReplayAction& action)
403   {
404     // Needs to be re-initialized for every action, hence here
405     double start_time = smpi_process()->simulated_elapsed();
406     args.parse(action, name);
407     kernel(action);
408     if (name != "Init")
409       log_timed_action(action, start_time);
410   }
411
412   virtual void kernel(simgrid::xbt::ReplayAction& action) = 0;
413
414   void* send_buffer(int size)
415   {
416     return smpi_get_tmp_sendbuffer(size);
417   }
418
419   void* recv_buffer(int size)
420   {
421     return smpi_get_tmp_recvbuffer(size);
422   }
423 };
424
425 class WaitAction : public ReplayAction<ActionArgParser> {
426 public:
427   WaitAction() : ReplayAction("Wait") {}
428   void kernel(simgrid::xbt::ReplayAction& action) override
429   {
430     std::string s = boost::algorithm::join(action, " ");
431     xbt_assert(get_reqq_self()->size(), "action wait not preceded by any irecv or isend: %s", s.c_str());
432     MPI_Request request = get_reqq_self()->back();
433     get_reqq_self()->pop_back();
434
435     if (request == nullptr) {
436       /* Assume that the trace is well formed, meaning the comm might have been caught by a MPI_test. Then just
437        * return.*/
438       return;
439     }
440
441     int rank = request->comm() != MPI_COMM_NULL ? request->comm()->rank() : -1;
442
443     // Must be taken before Request::wait() since the request may be set to
444     // MPI_REQUEST_NULL by Request::wait!
445     int src                  = request->comm()->group()->rank(request->src());
446     int dst                  = request->comm()->group()->rank(request->dst());
447     bool is_wait_for_receive = (request->flags() & RECV);
448     // TODO: Here we take the rank while we normally take the process id (look for my_proc_id)
449     TRACE_smpi_comm_in(rank, __func__, new simgrid::instr::NoOpTIData("wait"));
450
451     MPI_Status status;
452     Request::wait(&request, &status);
453
454     TRACE_smpi_comm_out(rank);
455     if (is_wait_for_receive)
456       TRACE_smpi_recv(src, dst, 0);
457   }
458 };
459
460 class SendAction : public ReplayAction<SendRecvParser> {
461 public:
462   SendAction() = delete;
463   SendAction(std::string name) : ReplayAction(name) {}
464   void kernel(simgrid::xbt::ReplayAction& action) override
465   {
466     int dst_traced = MPI_COMM_WORLD->group()->actor(args.partner)->getPid();
467
468     TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::Pt2PtTIData(name, args.partner, args.size,
469                                                                              Datatype::encode(args.datatype1)));
470     if (not TRACE_smpi_view_internals())
471       TRACE_smpi_send(my_proc_id, my_proc_id, dst_traced, 0, args.size * args.datatype1->size());
472
473     if (name == "send") {
474       Request::send(nullptr, args.size, args.datatype1, args.partner, 0, MPI_COMM_WORLD);
475     } else if (name == "Isend") {
476       MPI_Request request = Request::isend(nullptr, args.size, args.datatype1, args.partner, 0, MPI_COMM_WORLD);
477       get_reqq_self()->push_back(request);
478     } else {
479       xbt_die("Don't know this action, %s", name.c_str());
480     }
481
482     TRACE_smpi_comm_out(my_proc_id);
483   }
484 };
485
486 class RecvAction : public ReplayAction<SendRecvParser> {
487 public:
488   RecvAction() = delete;
489   explicit RecvAction(std::string name) : ReplayAction(name) {}
490   void kernel(simgrid::xbt::ReplayAction& action) override
491   {
492     int src_traced = MPI_COMM_WORLD->group()->actor(args.partner)->getPid();
493
494     TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::Pt2PtTIData(name, args.partner, args.size,
495                                                                              Datatype::encode(args.datatype1)));
496
497     MPI_Status status;
498     // unknown size from the receiver point of view
499     if (args.size <= 0.0) {
500       Request::probe(args.partner, 0, MPI_COMM_WORLD, &status);
501       args.size = status.count;
502     }
503
504     if (name == "recv") {
505       Request::recv(nullptr, args.size, args.datatype1, args.partner, 0, MPI_COMM_WORLD, &status);
506     } else if (name == "Irecv") {
507       MPI_Request request = Request::irecv(nullptr, args.size, args.datatype1, args.partner, 0, MPI_COMM_WORLD);
508       get_reqq_self()->push_back(request);
509     }
510
511     TRACE_smpi_comm_out(my_proc_id);
512     // TODO: Check why this was only activated in the "recv" case and not in the "Irecv" case
513     if (name == "recv" && not TRACE_smpi_view_internals()) {
514       TRACE_smpi_recv(src_traced, my_proc_id, 0);
515     }
516   }
517 };
518
519 class ComputeAction : public ReplayAction<ComputeParser> {
520 public:
521   ComputeAction() : ReplayAction("compute") {}
522   void kernel(simgrid::xbt::ReplayAction& action) override
523   {
524     TRACE_smpi_computing_in(my_proc_id, args.flops);
525     smpi_execute_flops(args.flops);
526     TRACE_smpi_computing_out(my_proc_id);
527   }
528 };
529
530 class TestAction : public ReplayAction<ActionArgParser> {
531 public:
532   TestAction() : ReplayAction("Test") {}
533   void kernel(simgrid::xbt::ReplayAction& action) override
534   {
535     MPI_Request request = get_reqq_self()->back();
536     get_reqq_self()->pop_back();
537     // if request is null here, this may mean that a previous test has succeeded
538     // Different times in traced application and replayed version may lead to this
539     // In this case, ignore the extra calls.
540     if (request != nullptr) {
541       TRACE_smpi_testing_in(my_proc_id);
542
543       MPI_Status status;
544       int flag = Request::test(&request, &status);
545
546       XBT_DEBUG("MPI_Test result: %d", flag);
547       /* push back request in vector to be caught by a subsequent wait. if the test did succeed, the request is now
548        * nullptr.*/
549       get_reqq_self()->push_back(request);
550
551       TRACE_smpi_testing_out(my_proc_id);
552     }
553   }
554 };
555
556 class InitAction : public ReplayAction<ActionArgParser> {
557 public:
558   InitAction() : ReplayAction("Init") {}
559   void kernel(simgrid::xbt::ReplayAction& action) override
560   {
561     CHECK_ACTION_PARAMS(action, 0, 1)
562     MPI_DEFAULT_TYPE = (action.size() > 2) ? MPI_DOUBLE // default MPE datatype
563                                            : MPI_BYTE;  // default TAU datatype
564
565     /* start a simulated timer */
566     smpi_process()->simulated_start();
567     /*initialize the number of active processes */
568     active_processes = smpi_process_count();
569
570     set_reqq_self(new std::vector<MPI_Request>);
571   }
572 };
573
574 class CommunicatorAction : public ReplayAction<ActionArgParser> {
575 public:
576   CommunicatorAction() : ReplayAction("Comm") {}
577   void kernel(simgrid::xbt::ReplayAction& action) override { /* nothing to do */}
578 };
579
580 class WaitAllAction : public ReplayAction<ActionArgParser> {
581 public:
582   WaitAllAction() : ReplayAction("waitAll") {}
583   void kernel(simgrid::xbt::ReplayAction& action) override
584   {
585     const unsigned int count_requests = get_reqq_self()->size();
586
587     if (count_requests > 0) {
588       TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::Pt2PtTIData("waitAll", -1, count_requests, ""));
589       std::vector<std::pair</*sender*/int,/*recv*/int>> sender_receiver;
590       for (const auto& req : (*get_reqq_self())) {
591         if (req && (req->flags() & RECV)) {
592           sender_receiver.push_back({req->src(), req->dst()});
593         }
594       }
595       MPI_Status status[count_requests];
596       Request::waitall(count_requests, &(*get_reqq_self())[0], status);
597
598       for (auto& pair : sender_receiver) {
599         TRACE_smpi_recv(pair.first, pair.second, 0);
600       }
601       TRACE_smpi_comm_out(my_proc_id);
602     }
603   }
604 };
605
606 class BarrierAction : public ReplayAction<ActionArgParser> {
607 public:
608   BarrierAction() : ReplayAction("barrier") {}
609   void kernel(simgrid::xbt::ReplayAction& action) override
610   {
611     TRACE_smpi_comm_in(my_proc_id, __func__, new simgrid::instr::NoOpTIData("barrier"));
612     Colls::barrier(MPI_COMM_WORLD);
613     TRACE_smpi_comm_out(my_proc_id);
614   }
615 };
616
617 class BcastAction : public ReplayAction<BcastArgParser> {
618 public:
619   BcastAction() : ReplayAction("bcast") {}
620   void kernel(simgrid::xbt::ReplayAction& action) override
621   {
622     TRACE_smpi_comm_in(my_proc_id, "action_bcast",
623                        new simgrid::instr::CollTIData("bcast", MPI_COMM_WORLD->group()->actor(args.root)->getPid(),
624                                                       -1.0, args.size, -1, Datatype::encode(args.datatype1), ""));
625
626     Colls::bcast(send_buffer(args.size * args.datatype1->size()), args.size, args.datatype1, args.root, MPI_COMM_WORLD);
627
628     TRACE_smpi_comm_out(my_proc_id);
629   }
630 };
631
632 class ReduceAction : public ReplayAction<ReduceArgParser> {
633 public:
634   ReduceAction() : ReplayAction("reduce") {}
635   void kernel(simgrid::xbt::ReplayAction& action) override
636   {
637     TRACE_smpi_comm_in(my_proc_id, "action_reduce",
638                        new simgrid::instr::CollTIData("reduce", MPI_COMM_WORLD->group()->actor(args.root)->getPid(), args.comp_size,
639                                                       args.comm_size, -1, Datatype::encode(args.datatype1), ""));
640
641     Colls::reduce(send_buffer(args.comm_size * args.datatype1->size()),
642         recv_buffer(args.comm_size * args.datatype1->size()), args.comm_size, args.datatype1, MPI_OP_NULL, args.root, MPI_COMM_WORLD);
643     smpi_execute_flops(args.comp_size);
644
645     TRACE_smpi_comm_out(my_proc_id);
646   }
647 };
648
649 class AllReduceAction : public ReplayAction<AllReduceArgParser> {
650 public:
651   AllReduceAction() : ReplayAction("allReduce") {}
652   void kernel(simgrid::xbt::ReplayAction& action) override
653   {
654     TRACE_smpi_comm_in(my_proc_id, "action_allReduce", new simgrid::instr::CollTIData("allReduce", -1, args.comp_size, args.comm_size, -1,
655                                                                                 Datatype::encode(args.datatype1), ""));
656
657     Colls::allreduce(send_buffer(args.comm_size * args.datatype1->size()),
658         recv_buffer(args.comm_size * args.datatype1->size()), args.comm_size, args.datatype1, MPI_OP_NULL, MPI_COMM_WORLD);
659     smpi_execute_flops(args.comp_size);
660
661     TRACE_smpi_comm_out(my_proc_id);
662   }
663 };
664
665 class AllToAllAction : public ReplayAction<AllToAllArgParser> {
666 public:
667   AllToAllAction() : ReplayAction("allToAll") {}
668   void kernel(simgrid::xbt::ReplayAction& action) override
669   {
670     TRACE_smpi_comm_in(my_proc_id, "action_allToAll",
671                      new simgrid::instr::CollTIData("allToAll", -1, -1.0, args.send_size, args.recv_size,
672                                                     Datatype::encode(args.datatype1),
673                                                     Datatype::encode(args.datatype2)));
674
675     Colls::alltoall(send_buffer(args.send_size*args.comm_size* args.datatype1->size()), 
676       args.send_size, args.datatype1, recv_buffer(args.recv_size * args.comm_size * args.datatype2->size()),
677       args.recv_size, args.datatype2, MPI_COMM_WORLD);
678
679     TRACE_smpi_comm_out(my_proc_id);
680   }
681 };
682
683 class GatherAction : public ReplayAction<GatherArgParser> {
684 public:
685   GatherAction(std::string name) : ReplayAction(name) {}
686   void kernel(simgrid::xbt::ReplayAction& action) override
687   {
688     TRACE_smpi_comm_in(my_proc_id, name.c_str(), new simgrid::instr::CollTIData(name, (name == "gather") ? args.root : -1, -1.0, args.send_size, args.recv_size,
689                                                                           Datatype::encode(args.datatype1), Datatype::encode(args.datatype2)));
690
691     if (name == "gather") {
692       int rank = MPI_COMM_WORLD->rank();
693       Colls::gather(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
694                  (rank == args.root) ? recv_buffer(args.recv_size * args.comm_size * args.datatype2->size()) : nullptr, args.recv_size, args.datatype2, args.root, MPI_COMM_WORLD);
695     }
696     else
697       Colls::allgather(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
698                        recv_buffer(args.recv_size * args.datatype2->size()), args.recv_size, args.datatype2, MPI_COMM_WORLD);
699
700     TRACE_smpi_comm_out(my_proc_id);
701   }
702 };
703
704 class GatherVAction : public ReplayAction<GatherVArgParser> {
705 public:
706   GatherVAction(std::string name) : ReplayAction(name) {}
707   void kernel(simgrid::xbt::ReplayAction& action) override
708   {
709     int rank = MPI_COMM_WORLD->rank();
710
711     TRACE_smpi_comm_in(my_proc_id, name.c_str(), new simgrid::instr::VarCollTIData(
712                                                name, (name == "gatherV") ? args.root : -1, args.send_size, nullptr, -1, args.recvcounts,
713                                                Datatype::encode(args.datatype1), Datatype::encode(args.datatype2)));
714
715     if (name == "gatherV") {
716       Colls::gatherv(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1, 
717                      (rank == args.root) ? recv_buffer(args.recv_size_sum  * args.datatype2->size()) : nullptr, args.recvcounts->data(), args.disps.data(), args.datatype2, args.root,
718                      MPI_COMM_WORLD);
719     }
720     else {
721       Colls::allgatherv(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1, 
722                         recv_buffer(args.recv_size_sum * args.datatype2->size()), args.recvcounts->data(), args.disps.data(), args.datatype2,
723                     MPI_COMM_WORLD);
724     }
725
726     TRACE_smpi_comm_out(my_proc_id);
727   }
728 };
729
730 class ScatterAction : public ReplayAction<ScatterArgParser> {
731 public:
732   ScatterAction() : ReplayAction("scatter") {}
733   void kernel(simgrid::xbt::ReplayAction& action) override
734   {
735     int rank = MPI_COMM_WORLD->rank();
736     TRACE_smpi_comm_in(my_proc_id, "action_scatter", new simgrid::instr::CollTIData(name, args.root, -1.0, args.send_size, args.recv_size,
737                                                                           Datatype::encode(args.datatype1),
738                                                                           Datatype::encode(args.datatype2)));
739
740     Colls::scatter(send_buffer(args.send_size * args.datatype1->size()), args.send_size, args.datatype1,
741                   (rank == args.root) ? recv_buffer(args.recv_size * args.datatype2->size()) : nullptr, args.recv_size, args.datatype2, args.root, MPI_COMM_WORLD);
742
743     TRACE_smpi_comm_out(my_proc_id);
744   }
745 };
746
747
748 class ScatterVAction : public ReplayAction<ScatterVArgParser> {
749 public:
750   ScatterVAction() : ReplayAction("scatterV") {}
751   void kernel(simgrid::xbt::ReplayAction& action) override
752   {
753     int rank = MPI_COMM_WORLD->rank();
754     TRACE_smpi_comm_in(my_proc_id, "action_scatterv", new simgrid::instr::VarCollTIData(name, args.root, -1, args.sendcounts, args.recv_size,
755           nullptr, Datatype::encode(args.datatype1),
756           Datatype::encode(args.datatype2)));
757
758     Colls::scatterv((rank == args.root) ? send_buffer(args.send_size_sum * args.datatype1->size()) : nullptr, args.sendcounts->data(), args.disps.data(), 
759         args.datatype1, recv_buffer(args.recv_size * args.datatype2->size()), args.recv_size, args.datatype2, args.root,
760         MPI_COMM_WORLD);
761
762     TRACE_smpi_comm_out(my_proc_id);
763   }
764 };
765
766 class ReduceScatterAction : public ReplayAction<ReduceScatterArgParser> {
767 public:
768   ReduceScatterAction() : ReplayAction("reduceScatter") {}
769   void kernel(simgrid::xbt::ReplayAction& action) override
770   {
771     TRACE_smpi_comm_in(my_proc_id, "action_reducescatter",
772                        new simgrid::instr::VarCollTIData("reduceScatter", -1, 0, nullptr, -1, args.recvcounts,
773                                                          std::to_string(args.comp_size), /* ugly hack to print comp_size */
774                                                          Datatype::encode(args.datatype1)));
775
776     Colls::reduce_scatter(send_buffer(args.recv_size_sum * args.datatype1->size()), recv_buffer(args.recv_size_sum * args.datatype1->size()), 
777                           args.recvcounts->data(), args.datatype1, MPI_OP_NULL, MPI_COMM_WORLD);
778
779     smpi_execute_flops(args.comp_size);
780     TRACE_smpi_comm_out(my_proc_id);
781   }
782 };
783
784 class AllToAllVAction : public ReplayAction<AllToAllVArgParser> {
785 public:
786   AllToAllVAction() : ReplayAction("allToAllV") {}
787   void kernel(simgrid::xbt::ReplayAction& action) override
788   {
789     TRACE_smpi_comm_in(my_proc_id, __func__,
790                        new simgrid::instr::VarCollTIData(
791                            "allToAllV", -1, args.send_size_sum, args.sendcounts, args.recv_size_sum, args.recvcounts,
792                            Datatype::encode(args.datatype1), Datatype::encode(args.datatype2)));
793
794     Colls::alltoallv(send_buffer(args.send_buf_size * args.datatype1->size()), args.sendcounts->data(), args.senddisps.data(), args.datatype1,
795                      recv_buffer(args.recv_buf_size * args.datatype2->size()), args.recvcounts->data(), args.recvdisps.data(), args.datatype2, MPI_COMM_WORLD);
796
797     TRACE_smpi_comm_out(my_proc_id);
798   }
799 };
800 } // Replay Namespace
801 }} // namespace simgrid::smpi
802
803 /** @brief Only initialize the replay, don't do it for real */
804 void smpi_replay_init(int* argc, char*** argv)
805 {
806   simgrid::smpi::Process::init(argc, argv);
807   smpi_process()->mark_as_initialized();
808   smpi_process()->set_replaying(true);
809
810   int my_proc_id = Actor::self()->getPid();
811   TRACE_smpi_init(my_proc_id);
812   TRACE_smpi_computing_init(my_proc_id);
813   TRACE_smpi_comm_in(my_proc_id, "smpi_replay_run_init", new simgrid::instr::NoOpTIData("init"));
814   TRACE_smpi_comm_out(my_proc_id);
815   xbt_replay_action_register("init", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::InitAction().execute(action); });
816   xbt_replay_action_register("finalize", [](simgrid::xbt::ReplayAction& action) { /* nothing to do */ });
817   xbt_replay_action_register("comm_size", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::CommunicatorAction().execute(action); });
818   xbt_replay_action_register("comm_split",[](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::CommunicatorAction().execute(action); });
819   xbt_replay_action_register("comm_dup",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::CommunicatorAction().execute(action); });
820
821   xbt_replay_action_register("send",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::SendAction("send").execute(action); });
822   xbt_replay_action_register("Isend", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::SendAction("Isend").execute(action); });
823   xbt_replay_action_register("recv",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::RecvAction("recv").execute(action); });
824   xbt_replay_action_register("Irecv", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::RecvAction("Irecv").execute(action); });
825   xbt_replay_action_register("test",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::TestAction().execute(action); });
826   xbt_replay_action_register("wait",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::WaitAction().execute(action); });
827   xbt_replay_action_register("waitAll", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::WaitAllAction().execute(action); });
828   xbt_replay_action_register("barrier", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::BarrierAction().execute(action); });
829   xbt_replay_action_register("bcast",   [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::BcastAction().execute(action); });
830   xbt_replay_action_register("reduce",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::ReduceAction().execute(action); });
831   xbt_replay_action_register("allReduce", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::AllReduceAction().execute(action); });
832   xbt_replay_action_register("allToAll", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::AllToAllAction().execute(action); });
833   xbt_replay_action_register("allToAllV", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::AllToAllVAction().execute(action); });
834   xbt_replay_action_register("gather",   [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::GatherAction("gather").execute(action); });
835   xbt_replay_action_register("scatter",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::ScatterAction().execute(action); });
836   xbt_replay_action_register("gatherV",  [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::GatherVAction("gatherV").execute(action); });
837   xbt_replay_action_register("scatterV", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::ScatterVAction().execute(action); });
838   xbt_replay_action_register("allGather", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::GatherAction("allGather").execute(action); });
839   xbt_replay_action_register("allGatherV", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::GatherVAction("allGatherV").execute(action); });
840   xbt_replay_action_register("reduceScatter", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::ReduceScatterAction().execute(action); });
841   xbt_replay_action_register("compute", [](simgrid::xbt::ReplayAction& action) { simgrid::smpi::Replay::ComputeAction().execute(action); });
842
843   //if we have a delayed start, sleep here.
844   if(*argc>2){
845     double value = xbt_str_parse_double((*argv)[2], "%s is not a double");
846     XBT_VERB("Delayed start for instance - Sleeping for %f flops ",value );
847     smpi_execute_flops(value);
848   } else {
849     //UGLY: force a context switch to be sure that all MSG_processes begin initialization
850     XBT_DEBUG("Force context switch by smpi_execute_flops  - Sleeping for 0.0 flops ");
851     smpi_execute_flops(0.0);
852   }
853 }
854
855 /** @brief actually run the replay after initialization */
856 void smpi_replay_main(int* argc, char*** argv)
857 {
858   simgrid::xbt::replay_runner(*argc, *argv);
859
860   /* and now, finalize everything */
861   /* One active process will stop. Decrease the counter*/
862   XBT_DEBUG("There are %zu elements in reqq[*]", get_reqq_self()->size());
863   if (not get_reqq_self()->empty()) {
864     unsigned int count_requests=get_reqq_self()->size();
865     MPI_Request requests[count_requests];
866     MPI_Status status[count_requests];
867     unsigned int i=0;
868
869     for (auto const& req : *get_reqq_self()) {
870       requests[i] = req;
871       i++;
872     }
873     simgrid::smpi::Request::waitall(count_requests, requests, status);
874   }
875   delete get_reqq_self();
876   active_processes--;
877
878   if(active_processes==0){
879     /* Last process alive speaking: end the simulated timer */
880     XBT_INFO("Simulation time %f", smpi_process()->simulated_elapsed());
881     smpi_free_replay_tmp_buffers();
882   }
883
884   TRACE_smpi_comm_in(Actor::self()->getPid(), "smpi_replay_run_finalize", new simgrid::instr::NoOpTIData("finalize"));
885
886   smpi_process()->finalize();
887
888   TRACE_smpi_comm_out(Actor::self()->getPid());
889   TRACE_smpi_finalize(Actor::self()->getPid());
890 }
891
892 /** @brief chain a replay initialization and a replay start */
893 void smpi_replay_run(int* argc, char*** argv)
894 {
895   smpi_replay_init(argc, argv);
896   smpi_replay_main(argc, argv);
897 }