Logo AND Algorithmique Numérique Distribuée

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