Logo AND Algorithmique Numérique Distribuée

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