Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Throw unimplemented.
[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   smpi_mpi_recv(NULL, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD, &status);
263
264 #ifdef HAVE_TRACING
265   TRACE_smpi_ptp_out(rank, src_traced, rank, __FUNCTION__);
266   TRACE_smpi_recv(rank, src_traced, rank);
267 #endif
268
269   log_timed_action (action, clock);
270 }
271
272 static void action_Irecv(const char *const *action)
273 {
274   int from = atoi(action[2]);
275   double size=parse_double(action[3]);
276   double clock = smpi_process_simulated_elapsed();
277   MPI_Request request;
278
279   if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
280   else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
281
282 #ifdef HAVE_TRACING
283   int rank = smpi_comm_rank(MPI_COMM_WORLD);
284   int src_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), from);
285   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
286   extra->type = TRACING_IRECV;
287   extra->send_size = size;
288   extra->src = src_traced;
289   extra->dst = rank;
290   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
291   TRACE_smpi_ptp_in(rank, src_traced, rank, __FUNCTION__, extra);
292 #endif
293
294   request = smpi_mpi_irecv(NULL, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD);
295
296 #ifdef HAVE_TRACING
297   TRACE_smpi_ptp_out(rank, src_traced, rank, __FUNCTION__);
298   request->recv = 1;
299 #endif
300   xbt_dynar_push(reqq[smpi_comm_rank(MPI_COMM_WORLD)],&request);
301
302   log_timed_action (action, clock);
303 }
304
305 static void action_wait(const char *const *action){
306   double clock = smpi_process_simulated_elapsed();
307   MPI_Request request;
308   MPI_Status status;
309
310   xbt_assert(xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]),
311       "action wait not preceded by any irecv or isend: %s",
312       xbt_str_join_array(action," "));
313   request = xbt_dynar_pop_as(reqq[smpi_comm_rank(MPI_COMM_WORLD)],MPI_Request);
314   xbt_assert(request != NULL, "found null request in reqq");
315 #ifdef HAVE_TRACING
316   int rank = request->comm != MPI_COMM_NULL
317       ? smpi_comm_rank(request->comm)
318       : -1;
319
320   MPI_Group group = smpi_comm_group(request->comm);
321   int src_traced = smpi_group_rank(group, request->src);
322   int dst_traced = smpi_group_rank(group, request->dst);
323   int is_wait_for_receive = request->recv;
324   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
325   extra->type = TRACING_WAIT;
326   TRACE_smpi_ptp_in(rank, src_traced, dst_traced, __FUNCTION__, extra);
327 #endif
328   smpi_mpi_wait(&request, &status);
329 #ifdef HAVE_TRACING
330   TRACE_smpi_ptp_out(rank, src_traced, dst_traced, __FUNCTION__);
331   if (is_wait_for_receive) {
332     TRACE_smpi_recv(rank, src_traced, dst_traced);
333   }
334 #endif
335
336   log_timed_action (action, clock);
337 }
338
339 static void action_waitall(const char *const *action){
340   double clock = smpi_process_simulated_elapsed();
341   int count_requests=0;
342   unsigned int i=0;
343
344   count_requests=xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
345
346   if (count_requests>0) {
347     MPI_Request requests[count_requests];
348     MPI_Status status[count_requests];
349
350     /*  The reqq is an array of dynars. Its index corresponds to the rank.
351      Thus each rank saves its own requests to the array request. */
352     xbt_dynar_foreach(reqq[smpi_comm_rank(MPI_COMM_WORLD)],i,requests[i]); 
353
354   #ifdef HAVE_TRACING
355    //save information from requests
356
357    xbt_dynar_t srcs = xbt_dynar_new(sizeof(int), NULL);
358    xbt_dynar_t dsts = xbt_dynar_new(sizeof(int), NULL);
359    xbt_dynar_t recvs = xbt_dynar_new(sizeof(int), NULL);
360    for (i = 0; i < count_requests; i++) {
361     if(requests[i]){
362       int *asrc = xbt_new(int, 1);
363       int *adst = xbt_new(int, 1);
364       int *arecv = xbt_new(int, 1);
365       *asrc = requests[i]->src;
366       *adst = requests[i]->dst;
367       *arecv = requests[i]->recv;
368       xbt_dynar_insert_at(srcs, i, asrc);
369       xbt_dynar_insert_at(dsts, i, adst);
370       xbt_dynar_insert_at(recvs, i, arecv);
371       xbt_free(asrc);
372       xbt_free(adst);
373       xbt_free(arecv);
374     }else {
375       int *t = xbt_new(int, 1);
376       xbt_dynar_insert_at(srcs, i, t);
377       xbt_dynar_insert_at(dsts, i, t);
378       xbt_dynar_insert_at(recvs, i, t);
379       xbt_free(t);
380     }
381    }
382    int rank_traced = smpi_process_index();
383    instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
384    extra->type = TRACING_WAITALL;
385    extra->send_size=count_requests;
386    TRACE_smpi_ptp_in(rank_traced, -1, -1, __FUNCTION__,extra);
387  #endif
388
389     smpi_mpi_waitall(count_requests, requests, status);
390
391   #ifdef HAVE_TRACING
392    for (i = 0; i < count_requests; i++) {
393     int src_traced, dst_traced, is_wait_for_receive;
394     xbt_dynar_get_cpy(srcs, i, &src_traced);
395     xbt_dynar_get_cpy(dsts, i, &dst_traced);
396     xbt_dynar_get_cpy(recvs, i, &is_wait_for_receive);
397     if (is_wait_for_receive) {
398       TRACE_smpi_recv(rank_traced, src_traced, dst_traced);
399     }
400    }
401    TRACE_smpi_ptp_out(rank_traced, -1, -1, __FUNCTION__);
402    //clean-up of dynars
403    xbt_dynar_free(&srcs);
404    xbt_dynar_free(&dsts);
405    xbt_dynar_free(&recvs);
406   #endif
407
408    xbt_dynar_free_container(&(reqq[smpi_comm_rank(MPI_COMM_WORLD)]));
409   }
410   log_timed_action (action, clock);
411 }
412
413 static void action_barrier(const char *const *action){
414   double clock = smpi_process_simulated_elapsed();
415 #ifdef HAVE_TRACING
416   int rank = smpi_comm_rank(MPI_COMM_WORLD);
417   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
418   extra->type = TRACING_BARRIER;
419   TRACE_smpi_collective_in(rank, -1, __FUNCTION__, extra);
420 #endif
421   smpi_mpi_barrier(MPI_COMM_WORLD);
422 #ifdef HAVE_TRACING
423   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
424 #endif
425
426   log_timed_action (action, clock);
427 }
428
429
430 static void action_bcast(const char *const *action)
431 {
432   double size = parse_double(action[2]);
433   double clock = smpi_process_simulated_elapsed();
434   int root=0;
435   /*
436    * Initialize MPI_CURRENT_TYPE in order to decrease
437    * the number of the checks
438    * */
439   MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;  
440
441   if(action[3]) {
442     root= atoi(action[3]);
443     if(action[4]) {
444       MPI_CURRENT_TYPE=decode_datatype(action[4]);   
445     }
446   }
447
448 #ifdef HAVE_TRACING
449   int rank = smpi_comm_rank(MPI_COMM_WORLD);
450   int root_traced = smpi_group_index(smpi_comm_group(MPI_COMM_WORLD), root);
451
452   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
453   extra->type = TRACING_BCAST;
454   extra->send_size = size;
455   extra->root = root_traced;
456   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
457   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__, extra);
458
459 #endif
460
461   mpi_coll_bcast_fun(NULL, size, MPI_CURRENT_TYPE, root, MPI_COMM_WORLD);
462 #ifdef HAVE_TRACING
463   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
464 #endif
465
466   log_timed_action (action, clock);
467 }
468
469 static void action_reduce(const char *const *action)
470 {
471   double comm_size = parse_double(action[2]);
472   double comp_size = parse_double(action[3]);
473   double clock = smpi_process_simulated_elapsed();
474   int root=0;
475   MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
476
477   if(action[4]) {
478     root= atoi(action[4]);
479     if(action[5]) {
480       MPI_CURRENT_TYPE=decode_datatype(action[5]);
481     }
482   }
483
484 #ifdef HAVE_TRACING
485   int rank = smpi_comm_rank(MPI_COMM_WORLD);
486   int root_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), root);
487   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
488   extra->type = TRACING_REDUCE;
489   extra->send_size = comm_size;
490   extra->comp_size = comp_size;
491   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
492   extra->root = root_traced;
493
494   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__,extra);
495 #endif
496    mpi_coll_reduce_fun(NULL, NULL, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, root, MPI_COMM_WORLD);
497    smpi_execute_flops(comp_size);
498 #ifdef HAVE_TRACING
499   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
500 #endif
501
502   log_timed_action (action, clock);
503 }
504
505 static void action_allReduce(const char *const *action) {
506   double comm_size = parse_double(action[2]);
507   double comp_size = parse_double(action[3]);
508
509   if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
510   else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
511
512   double clock = smpi_process_simulated_elapsed();
513 #ifdef HAVE_TRACING
514   int rank = smpi_comm_rank(MPI_COMM_WORLD);
515   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
516   extra->type = TRACING_ALLREDUCE;
517   extra->send_size = comm_size;
518   extra->comp_size = comp_size;
519   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
520
521   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
522 #endif
523   mpi_coll_reduce_fun(NULL, NULL, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, 0, MPI_COMM_WORLD);
524   smpi_execute_flops(comp_size);
525   mpi_coll_bcast_fun(NULL, comm_size, MPI_CURRENT_TYPE, 0, MPI_COMM_WORLD);
526 #ifdef HAVE_TRACING
527   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
528 #endif
529
530   log_timed_action (action, clock);
531 }
532
533 static void action_allToAll(const char *const *action) {
534   double clock = smpi_process_simulated_elapsed();
535   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
536   int send_size = parse_double(action[2]);
537   int recv_size = parse_double(action[3]);
538   MPI_Datatype MPI_CURRENT_TYPE2;
539
540   if(action[4]) {
541     MPI_CURRENT_TYPE=decode_datatype(action[4]);
542     MPI_CURRENT_TYPE2=decode_datatype(action[5]);
543   }
544   else {
545     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
546     MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
547   }
548   void *send = calloc(send_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE));  
549   void *recv = calloc(recv_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE2));  
550
551 #ifdef HAVE_TRACING
552   int rank = smpi_process_index();
553   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
554   extra->type = TRACING_ALLTOALL;
555   extra->send_size = send_size;
556   extra->recv_size = recv_size;
557   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
558   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
559
560   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
561 #endif
562
563   mpi_coll_alltoall_fun(send, send_size, MPI_CURRENT_TYPE, recv, recv_size, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
564
565 #ifdef HAVE_TRACING
566   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
567 #endif
568
569   log_timed_action (action, clock);
570   xbt_free(send);
571   xbt_free(recv);
572 }
573
574
575 static void action_gather(const char *const *action) {
576   /*
577  The structure of the gather action for the rank 0 (total 4 processes) 
578  is the following:   
579  0 gather 68 68 0 0 0
580
581   where: 
582   1) 68 is the sendcounts
583   2) 68 is the recvcounts
584   3) 0 is the root node
585   4) 0 is the send datatype id, see decode_datatype()
586   5) 0 is the recv datatype id, see decode_datatype()
587   */
588   double clock = smpi_process_simulated_elapsed();
589   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
590   int send_size = parse_double(action[2]);
591   int recv_size = parse_double(action[3]);
592   MPI_Datatype MPI_CURRENT_TYPE2;
593   if(action[5]) {
594     MPI_CURRENT_TYPE=decode_datatype(action[5]);
595     MPI_CURRENT_TYPE2=decode_datatype(action[6]);
596   } else {
597     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
598     MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
599   }
600   void *send = calloc(send_size, smpi_datatype_size(MPI_CURRENT_TYPE));
601   void *recv = NULL;
602
603   int root=atoi(action[4]);
604   int rank = smpi_process_index();
605
606   if(rank==root)
607     recv = calloc(recv_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
608
609 #ifdef HAVE_TRACING
610   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
611   extra->type = TRACING_GATHER;
612   extra->send_size = send_size;
613   extra->recv_size = recv_size;
614   extra->root = root;
615   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
616   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
617
618   TRACE_smpi_collective_in(rank, root, __FUNCTION__, extra);
619 #endif
620 smpi_mpi_gather(send, send_size, MPI_CURRENT_TYPE,
621                 recv, recv_size, MPI_CURRENT_TYPE2,
622                 root, MPI_COMM_WORLD);
623
624 #ifdef HAVE_TRACING
625   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
626 #endif
627
628   log_timed_action (action, clock);
629   xbt_free(send);
630   xbt_free(recv);
631 }
632
633
634
635 static void action_gatherv(const char *const *action) {
636   /*
637  The structure of the gatherv action for the rank 0 (total 4 processes)
638  is the following:
639  0 gather 68 68 10 10 10 0 0 0
640
641   where:
642   1) 68 is the sendcount
643   2) 68 10 10 10 is the recvcounts
644   3) 0 is the root node
645   4) 0 is the send datatype id, see decode_datatype()
646   5) 0 is the recv datatype id, see decode_datatype()
647   */
648   double clock = smpi_process_simulated_elapsed();
649   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
650   int send_size = parse_double(action[2]);
651   int *disps = xbt_new0(int, comm_size);
652   int *recvcounts = xbt_new0(int, comm_size);
653   int i=0,recv_sum=0;
654
655   MPI_Datatype MPI_CURRENT_TYPE2;
656   if(action[4+comm_size]) {
657     MPI_CURRENT_TYPE=decode_datatype(action[4+comm_size]);
658     MPI_CURRENT_TYPE2=decode_datatype(action[5+comm_size]);
659   } else {
660     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
661     MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
662   }
663   void *send = calloc(send_size, smpi_datatype_size(MPI_CURRENT_TYPE));
664   void *recv = NULL;
665   for(i=0;i<comm_size;i++) {
666     recvcounts[i] = atoi(action[i+3]);
667     recv_sum=recv_sum+recvcounts[i];
668     disps[i] = 0;
669   }
670
671   int root=atoi(action[3+comm_size]);
672   int rank = smpi_process_index();
673
674   if(rank==root)
675     recv = calloc(recv_sum, smpi_datatype_size(MPI_CURRENT_TYPE2));
676
677 #ifdef HAVE_TRACING
678   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
679   extra->type = TRACING_GATHERV;
680   extra->send_size = send_size;
681   extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
682   for(i=0; i< comm_size; i++)//copy data to avoid bad free
683     extra->recvcounts[i] = recvcounts[i];
684   extra->root = root;
685   extra->num_processes = comm_size;
686   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
687   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
688
689   TRACE_smpi_collective_in(rank, root, __FUNCTION__, extra);
690 #endif
691 smpi_mpi_gatherv(send, send_size, MPI_CURRENT_TYPE,
692                 recv, recvcounts, disps, MPI_CURRENT_TYPE2,
693                 root, MPI_COMM_WORLD);
694
695 #ifdef HAVE_TRACING
696   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
697 #endif
698
699   log_timed_action (action, clock);
700   xbt_free(recvcounts);
701   xbt_free(send);
702   xbt_free(recv);
703   xbt_free(disps);
704
705 }
706
707 static void action_reducescatter(const char *const *action) {
708
709     /*
710  The structure of the reducescatter action for the rank 0 (total 4 processes) 
711  is the following:   
712 0 reduceScatter 275427 275427 275427 204020 11346849 0
713
714   where: 
715   1) The first four values after the name of the action declare the recvcounts array
716   2) The value 11346849 is the amount of instructions
717   3) The last value corresponds to the datatype, see decode_datatype().
718
719   We analyze a MPI_Reduce_scatter call to one MPI_Reduce and one MPI_Scatterv.
720
721    */
722
723   double clock = smpi_process_simulated_elapsed();
724   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
725   int comp_size = parse_double(action[2+comm_size]);
726   int *recvcounts = xbt_new0(int, comm_size);  
727   int *disps = xbt_new0(int, comm_size);  
728   int i=0,recv_sum=0;
729   int root=0;
730   int rank = smpi_process_index();
731
732   if(action[3+comm_size])
733     MPI_CURRENT_TYPE=decode_datatype(action[3+comm_size]);
734   else
735     MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
736
737   for(i=0;i<comm_size;i++) {
738     recvcounts[i] = atoi(action[i+2]);
739     recv_sum=recv_sum+recvcounts[i];
740     disps[i] = 0;
741   }
742
743 #ifdef HAVE_TRACING
744   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
745   extra->type = TRACING_REDUCE_SCATTER;
746   extra->send_size = 0;
747   extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
748   for(i=0; i< comm_size; i++)//copy data to avoid bad free
749     extra->recvcounts[i] = recvcounts[i];
750   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
751   extra->comp_size = comp_size;
752   extra->num_processes = comm_size;
753
754
755   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
756 #endif
757    mpi_coll_reduce_fun(NULL, NULL, recv_sum, MPI_CURRENT_TYPE, MPI_OP_NULL,
758        root, MPI_COMM_WORLD);
759    smpi_mpi_scatterv(NULL, recvcounts, disps, MPI_CURRENT_TYPE, NULL,
760                       recvcounts[rank], MPI_CURRENT_TYPE, 0, MPI_COMM_WORLD);
761    smpi_execute_flops(comp_size);
762
763
764 #ifdef HAVE_TRACING
765   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
766 #endif
767   xbt_free(recvcounts);
768   xbt_free(disps);
769   log_timed_action (action, clock);
770 }
771
772
773 static void action_allgatherv(const char *const *action) {
774
775   /*
776  The structure of the allgatherv action for the rank 0 (total 4 processes) 
777  is the following:   
778 0 allGatherV 275427 275427 275427 275427 204020
779
780   where: 
781   1) 275427 is the sendcount
782   2) The next four elements declare the recvcounts array
783   3) No more values mean that the datatype for sent and receive buffer
784   is the default one, see decode_datatype().
785
786    */
787
788   double clock = smpi_process_simulated_elapsed();
789
790   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
791   int i=0;
792   int sendcount=atoi(action[2]);
793   int *recvcounts = xbt_new0(int, comm_size);  
794   int *disps = xbt_new0(int, comm_size);  
795   int recv_sum=0;  
796   MPI_Datatype MPI_CURRENT_TYPE2;
797
798   if(action[3+comm_size]) {
799     MPI_CURRENT_TYPE = decode_datatype(action[3+comm_size]);
800     MPI_CURRENT_TYPE2 = decode_datatype(action[4+comm_size]);
801   } else {
802     MPI_CURRENT_TYPE = MPI_DEFAULT_TYPE;
803     MPI_CURRENT_TYPE2 = MPI_DEFAULT_TYPE;    
804   }
805   void *sendbuf = calloc(sendcount, smpi_datatype_size(MPI_CURRENT_TYPE));    
806
807   for(i=0;i<comm_size;i++) {
808     recvcounts[i] = atoi(action[i+3]);
809     recv_sum=recv_sum+recvcounts[i];
810   }
811   void *recvbuf = calloc(recv_sum, smpi_datatype_size(MPI_CURRENT_TYPE2));  
812
813 #ifdef HAVE_TRACING
814   int rank = smpi_process_index();
815   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
816   extra->type = TRACING_ALLGATHERV;
817   extra->send_size = sendcount;
818   extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
819   for(i=0; i< comm_size; i++)//copy data to avoid bad free
820     extra->recvcounts[i] = recvcounts[i];
821   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
822   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
823   extra->num_processes = comm_size;
824
825   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
826 #endif
827
828 mpi_coll_allgatherv_fun(sendbuf, sendcount, MPI_CURRENT_TYPE, recvbuf, recvcounts, disps, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
829
830 #ifdef HAVE_TRACING
831   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
832 #endif
833
834   log_timed_action (action, clock);
835   xbt_free(sendbuf);
836   xbt_free(recvbuf);
837   xbt_free(recvcounts);
838   xbt_free(disps);
839 }
840
841
842 static void action_allToAllv(const char *const *action) {
843   /*
844  The structure of the allToAllV action for the rank 0 (total 4 processes) 
845  is the following:   
846   0 allToAllV 100 1 7 10 12 100 1 70 10 5
847
848   where: 
849   1) 100 is the size of the send buffer *sizeof(int),
850   2) 1 7 10 12 is the sendcounts array
851   3) 100*sizeof(int) is the size of the receiver buffer
852   4)  1 70 10 5 is the recvcounts array
853
854    */
855
856
857   double clock = smpi_process_simulated_elapsed();
858
859   int comm_size = smpi_comm_size(MPI_COMM_WORLD);
860   int send_buf_size=0,recv_buf_size=0,i=0;
861   int *sendcounts = xbt_new0(int, comm_size);  
862   int *recvcounts = xbt_new0(int, comm_size);  
863   int *senddisps = xbt_new0(int, comm_size);  
864   int *recvdisps = xbt_new0(int, comm_size);  
865
866   MPI_Datatype MPI_CURRENT_TYPE2;
867
868   send_buf_size=parse_double(action[2]);
869   recv_buf_size=parse_double(action[3+comm_size]);
870   if(action[4+2*comm_size]) {
871     MPI_CURRENT_TYPE=decode_datatype(action[4+2*comm_size]);
872     MPI_CURRENT_TYPE2=decode_datatype(action[5+2*comm_size]);
873   }
874   else {
875       MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
876       MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
877   }
878
879   void *sendbuf = calloc(send_buf_size, smpi_datatype_size(MPI_CURRENT_TYPE));  
880   void *recvbuf = calloc(recv_buf_size, smpi_datatype_size(MPI_CURRENT_TYPE2));  
881
882   for(i=0;i<comm_size;i++) {
883     sendcounts[i] = atoi(action[i+3]);
884     recvcounts[i] = atoi(action[i+4+comm_size]);
885   }
886
887
888 #ifdef HAVE_TRACING
889   int rank = smpi_process_index();
890   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
891   extra->type = TRACING_ALLTOALLV;
892   extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
893   extra->sendcounts= xbt_malloc(comm_size*sizeof(int));
894   extra->num_processes = comm_size;
895
896   for(i=0; i< comm_size; i++){//copy data to avoid bad free
897     extra->send_size += sendcounts[i];
898     extra->sendcounts[i] = sendcounts[i];
899     extra->recv_size += recvcounts[i];
900     extra->recvcounts[i] = recvcounts[i];
901   }
902   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
903   extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
904
905   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
906 #endif
907     mpi_coll_alltoallv_fun(sendbuf, sendcounts, senddisps, MPI_CURRENT_TYPE,
908                                recvbuf, recvcounts, recvdisps, MPI_CURRENT_TYPE,
909                                MPI_COMM_WORLD);
910 #ifdef HAVE_TRACING
911   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
912 #endif
913
914   log_timed_action (action, clock);
915   xbt_free(sendbuf);
916   xbt_free(recvbuf);
917   xbt_free(sendcounts);
918   xbt_free(recvcounts);
919   xbt_free(senddisps);
920   xbt_free(recvdisps);
921 }
922
923 void smpi_replay_init(int *argc, char***argv){
924   smpi_process_init(argc, argv);
925   smpi_process_mark_as_initialized();
926 #ifdef HAVE_TRACING
927   int rank = smpi_process_index();
928   TRACE_smpi_init(rank);
929   TRACE_smpi_computing_init(rank);
930   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
931   extra->type = TRACING_INIT;
932   TRACE_smpi_collective_in(rank, -1, __FUNCTION__, extra);
933   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
934 #endif
935
936   if (!smpi_process_index()){
937     _xbt_replay_action_init();
938     xbt_replay_action_register("init",       action_init);
939     xbt_replay_action_register("finalize",   action_finalize);
940     xbt_replay_action_register("comm_size",  action_comm_size);
941     xbt_replay_action_register("comm_split", action_comm_split);
942     xbt_replay_action_register("comm_dup",   action_comm_dup);
943     xbt_replay_action_register("send",       action_send);
944     xbt_replay_action_register("Isend",      action_Isend);
945     xbt_replay_action_register("recv",       action_recv);
946     xbt_replay_action_register("Irecv",      action_Irecv);
947     xbt_replay_action_register("wait",       action_wait);
948     xbt_replay_action_register("waitAll",    action_waitall);
949     xbt_replay_action_register("barrier",    action_barrier);
950     xbt_replay_action_register("bcast",      action_bcast);
951     xbt_replay_action_register("reduce",     action_reduce);
952     xbt_replay_action_register("allReduce",  action_allReduce);
953     xbt_replay_action_register("allToAll",   action_allToAll);
954     xbt_replay_action_register("allToAllV",  action_allToAllv);
955     xbt_replay_action_register("gather",  action_gather);
956     xbt_replay_action_register("gatherV",  action_gatherv);
957     xbt_replay_action_register("allGatherV",  action_allgatherv);
958     xbt_replay_action_register("reduceScatter",  action_reducescatter);
959     xbt_replay_action_register("compute",    action_compute);
960   }
961
962   xbt_replay_action_runner(*argc, *argv);
963 }
964
965 int smpi_replay_finalize(){
966   double sim_time= 1.;
967   /* One active process will stop. Decrease the counter*/
968   XBT_DEBUG("There are %lu elements in reqq[*]",
969             xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]));
970   if (!xbt_dynar_is_empty(reqq[smpi_comm_rank(MPI_COMM_WORLD)])){
971     int count_requests=xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
972     MPI_Request requests[count_requests];
973     MPI_Status status[count_requests];
974     unsigned int i;
975
976     xbt_dynar_foreach(reqq[smpi_comm_rank(MPI_COMM_WORLD)],i,requests[i]);
977     smpi_mpi_waitall(count_requests, requests, status);
978     active_processes--;
979   } else {
980     active_processes--;
981   }
982
983   xbt_dynar_free_container(&(reqq[smpi_comm_rank(MPI_COMM_WORLD)]));
984
985   if(!active_processes){
986     /* Last process alive speaking */
987     /* end the simulated timer */
988     sim_time = smpi_process_simulated_elapsed();
989     XBT_INFO("Simulation time %f", sim_time);
990     _xbt_replay_action_exit();
991     xbt_free(reqq);
992     reqq = NULL;
993   }
994   smpi_mpi_barrier(MPI_COMM_WORLD);
995 #ifdef HAVE_TRACING
996   int rank = smpi_process_index();
997   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
998   extra->type = TRACING_FINALIZE;
999   TRACE_smpi_collective_in(rank, -1, __FUNCTION__, extra);
1000 #endif
1001   smpi_process_finalize();
1002 #ifdef HAVE_TRACING
1003   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
1004   TRACE_smpi_finalize(smpi_process_index());
1005 #endif
1006   smpi_process_destroy();
1007   return MPI_SUCCESS;
1008 }