1 /* Copyright (c) 2009-2013. The SimGrid Team.
2 * All rights reserved. */
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. */
11 #include <xbt/replay.h>
13 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_replay,smpi,"Trace Replay with SMPI");
15 int communicator_size = 0;
16 static int active_processes = 0;
17 xbt_dynar_t *reqq = NULL;
19 MPI_Datatype MPI_DEFAULT_TYPE;
20 MPI_Datatype MPI_CURRENT_TYPE;
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);
31 xbt_dynar_t irecvs; /* of MPI_Request */
32 } s_smpi_replay_globals_t, *smpi_replay_globals_t;
36 static double parse_double(const char *string)
40 value = strtod(string, &endptr);
42 THROWF(unknown_error, 0, "%s is not a double", string);
46 static MPI_Datatype decode_datatype(const char *const action)
48 // Declared datatypes,
53 MPI_CURRENT_TYPE=MPI_DOUBLE;
56 MPI_CURRENT_TYPE=MPI_INT;
59 MPI_CURRENT_TYPE=MPI_CHAR;
62 MPI_CURRENT_TYPE=MPI_SHORT;
65 MPI_CURRENT_TYPE=MPI_LONG;
68 MPI_CURRENT_TYPE=MPI_FLOAT;
71 MPI_CURRENT_TYPE=MPI_BYTE;
74 MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
77 return MPI_CURRENT_TYPE;
81 const char* encode_datatype(MPI_Datatype datatype)
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){
89 if(datatype==MPI_DOUBLE)
93 if(datatype==MPI_CHAR)
95 if(datatype==MPI_SHORT)
97 if(datatype==MPI_LONG)
99 if(datatype==MPI_FLOAT)
102 // default - not implemented.
103 // do not warn here as we pass in this function even for other trace formats
107 static void action_init(const char *const *action)
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);
114 if(action[2]) MPI_DEFAULT_TYPE= MPI_DOUBLE; // default MPE dataype
115 else MPI_DEFAULT_TYPE= MPI_BYTE; // default TAU datatype
117 smpi_process_set_user_data((void*) globals);
119 /* start a simulated timer */
120 smpi_process_simulated_start();
121 /*initialize the number of active processes */
122 active_processes = smpi_process_count();
125 reqq=xbt_new0(xbt_dynar_t,active_processes);
127 for(i=0;i<active_processes;i++){
128 reqq[i]=xbt_dynar_new(sizeof(MPI_Request),NULL);
133 static void action_finalize(const char *const *action)
135 smpi_replay_globals_t globals =
136 (smpi_replay_globals_t) smpi_process_get_user_data();
138 XBT_DEBUG("There are %lu irecvs in the dynar",
139 xbt_dynar_length(globals->irecvs));
140 xbt_dynar_free_container(&(globals->irecvs));
145 static void action_comm_size(const char *const *action)
147 double clock = smpi_process_simulated_elapsed();
149 communicator_size = parse_double(action[2]);
150 log_timed_action (action, clock);
153 static void action_comm_split(const char *const *action)
155 double clock = smpi_process_simulated_elapsed();
157 log_timed_action (action, clock);
160 static void action_comm_dup(const char *const *action)
162 double clock = smpi_process_simulated_elapsed();
164 log_timed_action (action, clock);
167 static void action_compute(const char *const *action)
169 double clock = smpi_process_simulated_elapsed();
170 double flops= parse_double(action[2]);
172 int rank = smpi_comm_rank(MPI_COMM_WORLD);
173 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
174 extra->type=TRACING_COMPUTING;
175 extra->comp_size=flops;
176 TRACE_smpi_computing_in(rank, extra);
178 smpi_execute_flops(flops);
180 TRACE_smpi_computing_out(rank);
183 log_timed_action (action, clock);
186 static void action_send(const char *const *action)
188 int to = atoi(action[2]);
189 double size=parse_double(action[3]);
190 double clock = smpi_process_simulated_elapsed();
193 MPI_CURRENT_TYPE=decode_datatype(action[4]);
195 MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
199 int rank = smpi_comm_rank(MPI_COMM_WORLD);
201 int dst_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), to);
202 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
203 extra->type = TRACING_SEND;
204 extra->send_size = size;
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));
212 smpi_mpi_send(NULL, size, MPI_CURRENT_TYPE, to , 0, MPI_COMM_WORLD);
214 log_timed_action (action, clock);
217 TRACE_smpi_ptp_out(rank, rank, dst_traced, __FUNCTION__);
222 static void action_Isend(const char *const *action)
224 int to = atoi(action[2]);
225 double size=parse_double(action[3]);
226 double clock = smpi_process_simulated_elapsed();
229 if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
230 else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
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_new0(s_instr_extra_data_t,1);
236 extra->type = TRACING_ISEND;
237 extra->send_size = size;
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));
245 request = smpi_mpi_isend(NULL, size, MPI_CURRENT_TYPE, to, 0,MPI_COMM_WORLD);
248 TRACE_smpi_ptp_out(rank, rank, dst_traced, __FUNCTION__);
252 xbt_dynar_push(reqq[smpi_comm_rank(MPI_COMM_WORLD)],&request);
254 log_timed_action (action, clock);
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();
263 if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
264 else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
267 int rank = smpi_comm_rank(MPI_COMM_WORLD);
268 int src_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), from);
270 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
271 extra->type = TRACING_RECV;
272 extra->send_size = size;
273 extra->src = src_traced;
275 extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
276 TRACE_smpi_ptp_in(rank, src_traced, rank, __FUNCTION__, extra);
279 smpi_mpi_recv(NULL, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD, &status);
282 TRACE_smpi_ptp_out(rank, src_traced, rank, __FUNCTION__);
283 TRACE_smpi_recv(rank, src_traced, rank);
286 log_timed_action (action, clock);
289 static void action_Irecv(const char *const *action)
291 int from = atoi(action[2]);
292 double size=parse_double(action[3]);
293 double clock = smpi_process_simulated_elapsed();
296 smpi_replay_globals_t globals =
297 (smpi_replay_globals_t) smpi_process_get_user_data();
299 if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
300 else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
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_new0(s_instr_extra_data_t,1);
306 extra->type = TRACING_IRECV;
307 extra->send_size = size;
308 extra->src = src_traced;
310 extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
311 TRACE_smpi_ptp_in(rank, src_traced, rank, __FUNCTION__, extra);
314 request = smpi_mpi_irecv(NULL, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD);
317 TRACE_smpi_ptp_out(rank, src_traced, rank, __FUNCTION__);
320 xbt_dynar_push(globals->irecvs,&request);
321 xbt_dynar_push(reqq[smpi_comm_rank(MPI_COMM_WORLD)],&request);
323 log_timed_action (action, clock);
326 static void action_wait(const char *const *action){
327 double clock = smpi_process_simulated_elapsed();
330 smpi_replay_globals_t globals =
331 (smpi_replay_globals_t) smpi_process_get_user_data();
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 xbt_assert(request != NULL, "found null request in globals->irecv");
339 int rank = request->comm != MPI_COMM_NULL
340 ? smpi_comm_rank(request->comm)
343 MPI_Group group = smpi_comm_group(request->comm);
344 int src_traced = smpi_group_rank(group, request->src);
345 int dst_traced = smpi_group_rank(group, request->dst);
346 int is_wait_for_receive = request->recv;
347 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
348 extra->type = TRACING_WAIT;
349 TRACE_smpi_ptp_in(rank, src_traced, dst_traced, __FUNCTION__, extra);
351 smpi_mpi_wait(&request, &status);
353 TRACE_smpi_ptp_out(rank, src_traced, dst_traced, __FUNCTION__);
354 if (is_wait_for_receive) {
355 TRACE_smpi_recv(rank, src_traced, dst_traced);
359 log_timed_action (action, clock);
362 static void action_waitall(const char *const *action){
363 double clock = smpi_process_simulated_elapsed();
364 int count_requests=0;
367 count_requests=xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
369 if (count_requests>0) {
370 MPI_Request requests[count_requests];
371 MPI_Status status[count_requests];
373 /* The reqq is an array of dynars. Its index corresponds to the rank.
374 Thus each rank saves its own requests to the array request. */
375 xbt_dynar_foreach(reqq[smpi_comm_rank(MPI_COMM_WORLD)],i,requests[i]);
378 //save information from requests
380 xbt_dynar_t srcs = xbt_dynar_new(sizeof(int), NULL);
381 xbt_dynar_t dsts = xbt_dynar_new(sizeof(int), NULL);
382 xbt_dynar_t recvs = xbt_dynar_new(sizeof(int), NULL);
383 for (i = 0; i < count_requests; i++) {
385 int *asrc = xbt_new(int, 1);
386 int *adst = xbt_new(int, 1);
387 int *arecv = xbt_new(int, 1);
388 *asrc = requests[i]->src;
389 *adst = requests[i]->dst;
390 *arecv = requests[i]->recv;
391 xbt_dynar_insert_at(srcs, i, asrc);
392 xbt_dynar_insert_at(dsts, i, adst);
393 xbt_dynar_insert_at(recvs, i, arecv);
398 int *t = xbt_new(int, 1);
399 xbt_dynar_insert_at(srcs, i, t);
400 xbt_dynar_insert_at(dsts, i, t);
401 xbt_dynar_insert_at(recvs, i, t);
405 int rank_traced = smpi_process_index();
406 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
407 extra->type = TRACING_WAITALL;
408 extra->send_size=count_requests;
409 TRACE_smpi_ptp_in(rank_traced, -1, -1, __FUNCTION__,extra);
412 smpi_mpi_waitall(count_requests, requests, status);
415 for (i = 0; i < count_requests; i++) {
416 int src_traced, dst_traced, is_wait_for_receive;
417 xbt_dynar_get_cpy(srcs, i, &src_traced);
418 xbt_dynar_get_cpy(dsts, i, &dst_traced);
419 xbt_dynar_get_cpy(recvs, i, &is_wait_for_receive);
420 if (is_wait_for_receive) {
421 TRACE_smpi_recv(rank_traced, src_traced, dst_traced);
424 TRACE_smpi_ptp_out(rank_traced, -1, -1, __FUNCTION__);
426 xbt_dynar_free(&srcs);
427 xbt_dynar_free(&dsts);
428 xbt_dynar_free(&recvs);
431 xbt_dynar_reset(reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
433 log_timed_action (action, clock);
436 static void action_barrier(const char *const *action){
437 double clock = smpi_process_simulated_elapsed();
439 int rank = smpi_comm_rank(MPI_COMM_WORLD);
440 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
441 extra->type = TRACING_BARRIER;
442 TRACE_smpi_collective_in(rank, -1, __FUNCTION__, extra);
444 smpi_mpi_barrier(MPI_COMM_WORLD);
446 TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
449 log_timed_action (action, clock);
453 static void action_bcast(const char *const *action)
455 double size = parse_double(action[2]);
456 double clock = smpi_process_simulated_elapsed();
459 * Initialize MPI_CURRENT_TYPE in order to decrease
460 * the number of the checks
462 MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
465 root= atoi(action[3]);
467 MPI_CURRENT_TYPE=decode_datatype(action[4]);
472 int rank = smpi_comm_rank(MPI_COMM_WORLD);
473 int root_traced = smpi_group_index(smpi_comm_group(MPI_COMM_WORLD), root);
475 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
476 extra->type = TRACING_BCAST;
477 extra->send_size = size;
478 extra->root = root_traced;
479 extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
480 TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__, extra);
484 mpi_coll_bcast_fun(NULL, size, MPI_CURRENT_TYPE, root, MPI_COMM_WORLD);
486 TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
489 log_timed_action (action, clock);
492 static void action_reduce(const char *const *action)
494 double comm_size = parse_double(action[2]);
495 double comp_size = parse_double(action[3]);
496 double clock = smpi_process_simulated_elapsed();
498 MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
501 root= atoi(action[4]);
503 MPI_CURRENT_TYPE=decode_datatype(action[5]);
508 int rank = smpi_comm_rank(MPI_COMM_WORLD);
509 int root_traced = smpi_group_rank(smpi_comm_group(MPI_COMM_WORLD), root);
510 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
511 extra->type = TRACING_REDUCE;
512 extra->send_size = comm_size;
513 extra->comp_size = comp_size;
514 extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
515 extra->root = root_traced;
517 TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__,extra);
519 mpi_coll_reduce_fun(NULL, NULL, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, root, MPI_COMM_WORLD);
520 smpi_execute_flops(comp_size);
522 TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
525 log_timed_action (action, clock);
528 static void action_allReduce(const char *const *action) {
529 double comm_size = parse_double(action[2]);
530 double comp_size = parse_double(action[3]);
532 if(action[4]) MPI_CURRENT_TYPE=decode_datatype(action[4]);
533 else MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
535 double clock = smpi_process_simulated_elapsed();
537 int rank = smpi_comm_rank(MPI_COMM_WORLD);
538 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
539 extra->type = TRACING_ALLREDUCE;
540 extra->send_size = comm_size;
541 extra->comp_size = comp_size;
542 extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
544 TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
546 mpi_coll_reduce_fun(NULL, NULL, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, 0, MPI_COMM_WORLD);
547 smpi_execute_flops(comp_size);
548 mpi_coll_bcast_fun(NULL, comm_size, MPI_CURRENT_TYPE, 0, MPI_COMM_WORLD);
550 TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
553 log_timed_action (action, clock);
556 static void action_allToAll(const char *const *action) {
557 double clock = smpi_process_simulated_elapsed();
558 int comm_size = smpi_comm_size(MPI_COMM_WORLD);
559 int send_size = parse_double(action[2]);
560 int recv_size = parse_double(action[3]);
561 MPI_Datatype MPI_CURRENT_TYPE2;
564 MPI_CURRENT_TYPE=decode_datatype(action[4]);
565 MPI_CURRENT_TYPE2=decode_datatype(action[5]);
568 MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
569 MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
571 void *send = calloc(send_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE));
572 void *recv = calloc(recv_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
575 int rank = smpi_process_index();
576 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
577 extra->type = TRACING_ALLTOALL;
578 extra->send_size = send_size;
579 extra->recv_size = recv_size;
580 extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
581 extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
583 TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
586 mpi_coll_alltoall_fun(send, send_size, MPI_CURRENT_TYPE, recv, recv_size, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
589 TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
592 log_timed_action (action, clock);
598 static void action_gather(const char *const *action) {
600 The structure of the gather action for the rank 0 (total 4 processes)
605 1) 68 is the sendcounts
606 2) 68 is the recvcounts
607 3) 0 is the root node
608 4) 0 is the send datatype id, see decode_datatype()
609 5) 0 is the recv datatype id, see decode_datatype()
611 double clock = smpi_process_simulated_elapsed();
612 int comm_size = smpi_comm_size(MPI_COMM_WORLD);
613 int send_size = parse_double(action[2]);
614 int recv_size = parse_double(action[3]);
615 MPI_Datatype MPI_CURRENT_TYPE2;
617 MPI_CURRENT_TYPE=decode_datatype(action[5]);
618 MPI_CURRENT_TYPE2=decode_datatype(action[6]);
620 MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
621 MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
623 void *send = calloc(send_size, smpi_datatype_size(MPI_CURRENT_TYPE));
626 int root=atoi(action[4]);
627 int rank = smpi_process_index();
630 recv = calloc(recv_size*comm_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
633 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
634 extra->type = TRACING_GATHER;
635 extra->send_size = send_size;
636 extra->recv_size = recv_size;
638 extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
639 extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
641 TRACE_smpi_collective_in(rank, root, __FUNCTION__, extra);
643 smpi_mpi_gather(send, send_size, MPI_CURRENT_TYPE,
644 recv, recv_size, MPI_CURRENT_TYPE2,
645 root, MPI_COMM_WORLD);
648 TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
651 log_timed_action (action, clock);
658 static void action_gatherv(const char *const *action) {
660 The structure of the gatherv action for the rank 0 (total 4 processes)
662 0 gather 68 68 10 10 10 0 0 0
665 1) 68 is the sendcount
666 2) 68 10 10 10 is the recvcounts
667 3) 0 is the root node
668 4) 0 is the send datatype id, see decode_datatype()
669 5) 0 is the recv datatype id, see decode_datatype()
671 double clock = smpi_process_simulated_elapsed();
672 int comm_size = smpi_comm_size(MPI_COMM_WORLD);
673 int send_size = parse_double(action[2]);
674 int *disps = xbt_new0(int, comm_size);
675 int *recvcounts = xbt_new0(int, comm_size);
678 MPI_Datatype MPI_CURRENT_TYPE2;
679 if(action[4+comm_size]) {
680 MPI_CURRENT_TYPE=decode_datatype(action[4+comm_size]);
681 MPI_CURRENT_TYPE2=decode_datatype(action[5+comm_size]);
683 MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
684 MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
686 void *send = calloc(send_size, smpi_datatype_size(MPI_CURRENT_TYPE));
688 for(i=0;i<comm_size;i++) {
689 recvcounts[i] = atoi(action[i+3]);
690 recv_sum=recv_sum+recvcounts[i];
694 int root=atoi(action[3+comm_size]);
695 int rank = smpi_process_index();
698 recv = calloc(recv_sum, smpi_datatype_size(MPI_CURRENT_TYPE2));
701 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
702 extra->type = TRACING_GATHERV;
703 extra->send_size = send_size;
704 extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
705 for(i=0; i< comm_size; i++)//copy data to avoid bad free
706 extra->recvcounts[i] = recvcounts[i];
708 extra->num_processes = comm_size;
709 extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
710 extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
712 TRACE_smpi_collective_in(rank, root, __FUNCTION__, extra);
714 smpi_mpi_gatherv(send, send_size, MPI_CURRENT_TYPE,
715 recv, recvcounts, disps, MPI_CURRENT_TYPE2,
716 root, MPI_COMM_WORLD);
719 TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
722 log_timed_action (action, clock);
723 xbt_free(recvcounts);
730 static void action_reducescatter(const char *const *action) {
733 The structure of the reducescatter action for the rank 0 (total 4 processes)
735 0 reduceScatter 275427 275427 275427 204020 11346849 0
738 1) The first four values after the name of the action declare the recvcounts array
739 2) The value 11346849 is the amount of instructions
740 3) The last value corresponds to the datatype, see decode_datatype().
742 We analyze a MPI_Reduce_scatter call to one MPI_Reduce and one MPI_Scatterv.
746 double clock = smpi_process_simulated_elapsed();
747 int comm_size = smpi_comm_size(MPI_COMM_WORLD);
748 int comp_size = parse_double(action[2+comm_size]);
749 int *recvcounts = xbt_new0(int, comm_size);
750 int *disps = xbt_new0(int, comm_size);
753 int rank = smpi_process_index();
755 if(action[3+comm_size])
756 MPI_CURRENT_TYPE=decode_datatype(action[3+comm_size]);
758 MPI_CURRENT_TYPE= MPI_DEFAULT_TYPE;
760 for(i=0;i<comm_size;i++) {
761 recvcounts[i] = atoi(action[i+2]);
762 recv_sum=recv_sum+recvcounts[i];
767 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
768 extra->type = TRACING_REDUCE_SCATTER;
769 extra->send_size = 0;
770 extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
771 for(i=0; i< comm_size; i++)//copy data to avoid bad free
772 extra->recvcounts[i] = recvcounts[i];
773 extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
774 extra->comp_size = comp_size;
775 extra->num_processes = comm_size;
778 TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
780 mpi_coll_reduce_fun(NULL, NULL, recv_sum, MPI_CURRENT_TYPE, MPI_OP_NULL,
781 root, MPI_COMM_WORLD);
782 smpi_mpi_scatterv(NULL, recvcounts, disps, MPI_CURRENT_TYPE, NULL,
783 recvcounts[rank], MPI_CURRENT_TYPE, 0, MPI_COMM_WORLD);
784 smpi_execute_flops(comp_size);
788 TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
790 xbt_free(recvcounts);
792 log_timed_action (action, clock);
796 static void action_allgatherv(const char *const *action) {
799 The structure of the allgatherv action for the rank 0 (total 4 processes)
801 0 allGatherV 275427 275427 275427 275427 204020
804 1) 275427 is the sendcount
805 2) The next four elements declare the recvcounts array
806 3) No more values mean that the datatype for sent and receive buffer
807 is the default one, see decode_datatype().
811 double clock = smpi_process_simulated_elapsed();
813 int comm_size = smpi_comm_size(MPI_COMM_WORLD);
815 int sendcount=atoi(action[2]);
816 int *recvcounts = xbt_new0(int, comm_size);
817 int *disps = xbt_new0(int, comm_size);
819 MPI_Datatype MPI_CURRENT_TYPE2;
821 if(action[3+comm_size]) {
822 MPI_CURRENT_TYPE = decode_datatype(action[3+comm_size]);
823 MPI_CURRENT_TYPE2 = decode_datatype(action[4+comm_size]);
825 MPI_CURRENT_TYPE = MPI_DEFAULT_TYPE;
826 MPI_CURRENT_TYPE2 = MPI_DEFAULT_TYPE;
828 void *sendbuf = calloc(sendcount, smpi_datatype_size(MPI_CURRENT_TYPE));
830 for(i=0;i<comm_size;i++) {
831 recvcounts[i] = atoi(action[i+3]);
832 recv_sum=recv_sum+recvcounts[i];
834 void *recvbuf = calloc(recv_sum, smpi_datatype_size(MPI_CURRENT_TYPE2));
837 int rank = smpi_process_index();
838 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
839 extra->type = TRACING_ALLGATHERV;
840 extra->send_size = sendcount;
841 extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
842 for(i=0; i< comm_size; i++)//copy data to avoid bad free
843 extra->recvcounts[i] = recvcounts[i];
844 extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
845 extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
846 extra->num_processes = comm_size;
848 TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
851 mpi_coll_allgatherv_fun(sendbuf, sendcount, MPI_CURRENT_TYPE, recvbuf, recvcounts, disps, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
854 TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
857 log_timed_action (action, clock);
860 xbt_free(recvcounts);
865 static void action_allToAllv(const char *const *action) {
867 The structure of the allToAllV action for the rank 0 (total 4 processes)
869 0 allToAllV 100 1 7 10 12 100 1 70 10 5
872 1) 100 is the size of the send buffer *sizeof(int),
873 2) 1 7 10 12 is the sendcounts array
874 3) 100*sizeof(int) is the size of the receiver buffer
875 4) 1 70 10 5 is the recvcounts array
880 double clock = smpi_process_simulated_elapsed();
882 int comm_size = smpi_comm_size(MPI_COMM_WORLD);
883 int send_buf_size=0,recv_buf_size=0,i=0;
884 int *sendcounts = xbt_new0(int, comm_size);
885 int *recvcounts = xbt_new0(int, comm_size);
886 int *senddisps = xbt_new0(int, comm_size);
887 int *recvdisps = xbt_new0(int, comm_size);
889 MPI_Datatype MPI_CURRENT_TYPE2;
891 send_buf_size=parse_double(action[2]);
892 recv_buf_size=parse_double(action[3+comm_size]);
893 if(action[4+2*comm_size]) {
894 MPI_CURRENT_TYPE=decode_datatype(action[4+2*comm_size]);
895 MPI_CURRENT_TYPE2=decode_datatype(action[5+2*comm_size]);
898 MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
899 MPI_CURRENT_TYPE2=MPI_DEFAULT_TYPE;
902 void *sendbuf = calloc(send_buf_size, smpi_datatype_size(MPI_CURRENT_TYPE));
903 void *recvbuf = calloc(recv_buf_size, smpi_datatype_size(MPI_CURRENT_TYPE2));
905 for(i=0;i<comm_size;i++) {
906 sendcounts[i] = atoi(action[i+3]);
907 recvcounts[i] = atoi(action[i+4+comm_size]);
912 int rank = smpi_process_index();
913 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
914 extra->type = TRACING_ALLTOALLV;
915 extra->recvcounts= xbt_malloc(comm_size*sizeof(int));
916 extra->sendcounts= xbt_malloc(comm_size*sizeof(int));
917 extra->num_processes = comm_size;
919 for(i=0; i< comm_size; i++){//copy data to avoid bad free
920 extra->send_size += sendcounts[i];
921 extra->sendcounts[i] = sendcounts[i];
922 extra->recv_size += recvcounts[i];
923 extra->recvcounts[i] = recvcounts[i];
925 extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
926 extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
928 TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
930 mpi_coll_alltoallv_fun(sendbuf, sendcounts, senddisps, MPI_CURRENT_TYPE,
931 recvbuf, recvcounts, recvdisps, MPI_CURRENT_TYPE,
934 TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
937 log_timed_action (action, clock);
940 xbt_free(sendcounts);
941 xbt_free(recvcounts);
946 void smpi_replay_init(int *argc, char***argv){
947 smpi_process_init(argc, argv);
948 smpi_process_mark_as_initialized();
950 int rank = smpi_process_index();
951 TRACE_smpi_init(rank);
952 TRACE_smpi_computing_init(rank);
953 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
954 extra->type = TRACING_INIT;
955 TRACE_smpi_collective_in(rank, -1, __FUNCTION__, extra);
956 TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
959 if (!smpi_process_index()){
960 _xbt_replay_action_init();
961 xbt_replay_action_register("init", action_init);
962 xbt_replay_action_register("finalize", action_finalize);
963 xbt_replay_action_register("comm_size", action_comm_size);
964 xbt_replay_action_register("comm_split", action_comm_split);
965 xbt_replay_action_register("comm_dup", action_comm_dup);
966 xbt_replay_action_register("send", action_send);
967 xbt_replay_action_register("Isend", action_Isend);
968 xbt_replay_action_register("recv", action_recv);
969 xbt_replay_action_register("Irecv", action_Irecv);
970 xbt_replay_action_register("wait", action_wait);
971 xbt_replay_action_register("waitAll", action_waitall);
972 xbt_replay_action_register("barrier", action_barrier);
973 xbt_replay_action_register("bcast", action_bcast);
974 xbt_replay_action_register("reduce", action_reduce);
975 xbt_replay_action_register("allReduce", action_allReduce);
976 xbt_replay_action_register("allToAll", action_allToAll);
977 xbt_replay_action_register("allToAllV", action_allToAllv);
978 xbt_replay_action_register("gather", action_gather);
979 xbt_replay_action_register("gatherV", action_gatherv);
980 xbt_replay_action_register("allGatherV", action_allgatherv);
981 xbt_replay_action_register("reduceScatter", action_reducescatter);
982 xbt_replay_action_register("compute", action_compute);
985 xbt_replay_action_runner(*argc, *argv);
988 int smpi_replay_finalize(){
990 /* One active process will stop. Decrease the counter*/
992 XBT_DEBUG("There are %lu elements in reqq[*]",
993 xbt_dynar_length(reqq[smpi_comm_rank(MPI_COMM_WORLD)]));
994 xbt_dynar_free(&reqq[smpi_comm_rank(MPI_COMM_WORLD)]);
995 if(!active_processes){
996 /* Last process alive speaking */
997 /* end the simulated timer */
998 sim_time = smpi_process_simulated_elapsed();
999 XBT_INFO("Simulation time %g", sim_time);
1000 _xbt_replay_action_exit();
1004 smpi_mpi_barrier(MPI_COMM_WORLD);
1006 int rank = smpi_process_index();
1007 instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
1008 extra->type = TRACING_FINALIZE;
1009 TRACE_smpi_collective_in(rank, -1, __FUNCTION__, extra);
1011 smpi_process_finalize();
1013 TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
1014 TRACE_smpi_finalize(smpi_process_index());
1016 smpi_process_destroy();