//allocate a single buffer for all sends, growing it if needed
void* smpi_get_tmp_sendbuffer(int size){
- if (!_xbt_replay_is_active())
+ if (!smpi_process_get_replaying())
return xbt_malloc(size);
if (sendbuffer_size<size){
sendbuffer=xbt_realloc(sendbuffer,size);
}
//allocate a single buffer for all recv
void* smpi_get_tmp_recvbuffer(int size){
- if (!_xbt_replay_is_active())
+ if (!smpi_process_get_replaying())
return xbt_malloc(size);
if (recvbuffer_size<size){
recvbuffer=xbt_realloc(recvbuffer,size);
}
void smpi_free_tmp_buffer(void* buf){
- if (!_xbt_replay_is_active())
+ if (!smpi_process_get_replaying())
xbt_free(buf);
}
int flag = TRUE;
request = xbt_dynar_pop_as(reqq[smpi_process_index()],MPI_Request);
- xbt_assert(request != NULL, "found null request in reqq");
-
+ //if request is null here, this may mean that a previous test has succeeded
+ //Different times in traced application and replayed version may lead to this
+ //In this case, ignore the extra calls.
+ if(request){
#ifdef HAVE_TRACING
int rank = smpi_process_index();
instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
extra->type=TRACING_TEST;
TRACE_smpi_testing_in(rank, extra);
#endif
- flag = smpi_mpi_test(&request, &status);
+
+ flag = smpi_mpi_test(&request, &status);
+
XBT_DEBUG("MPI_Test result: %d", flag);
/* push back request in dynar to be caught by a subsequent wait. if the test
* did succeed, the request is now NULL.
#ifdef HAVE_TRACING
TRACE_smpi_testing_out(rank);
#endif
-
+ }
log_timed_action (action, clock);
}
xbt_dynar_free(&recvs);
#endif
- xbt_dynar_free_container(&(reqq[smpi_process_index()]));
+ int freedrank=smpi_process_index();
+ xbt_dynar_free_container(&(reqq[freedrank]));
+ reqq[freedrank]=xbt_dynar_new(sizeof(MPI_Request),&xbt_free_ref);
}
log_timed_action (action, clock);
}
TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__, extra);
#endif
-
- mpi_coll_bcast_fun(NULL, size, MPI_CURRENT_TYPE, root, MPI_COMM_WORLD);
+ void *sendbuf = smpi_get_tmp_sendbuffer(size* smpi_datatype_size(MPI_CURRENT_TYPE));
+ mpi_coll_bcast_fun(sendbuf, size, MPI_CURRENT_TYPE, root, MPI_COMM_WORLD);
#ifdef HAVE_TRACING
TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
#endif
-
log_timed_action (action, clock);
}
MPI_CURRENT_TYPE=decode_datatype(action[5]);
}
}
+
+
#ifdef HAVE_TRACING
int rank = smpi_process_index();
TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__,extra);
#endif
- mpi_coll_reduce_fun(NULL, NULL, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, root, MPI_COMM_WORLD);
+ void *recvbuf = smpi_get_tmp_sendbuffer(comm_size* smpi_datatype_size(MPI_CURRENT_TYPE));
+ void *sendbuf = smpi_get_tmp_sendbuffer(comm_size* smpi_datatype_size(MPI_CURRENT_TYPE));
+ mpi_coll_reduce_fun(sendbuf, recvbuf, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, root, MPI_COMM_WORLD);
smpi_execute_flops(comp_size);
#ifdef HAVE_TRACING
TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
#endif
-
log_timed_action (action, clock);
}
TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
#endif
- mpi_coll_allreduce_fun(NULL, NULL, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, MPI_COMM_WORLD);
+ void *recvbuf = smpi_get_tmp_sendbuffer(comm_size* smpi_datatype_size(MPI_CURRENT_TYPE));
+ void *sendbuf = smpi_get_tmp_sendbuffer(comm_size* smpi_datatype_size(MPI_CURRENT_TYPE));
+ mpi_coll_allreduce_fun(sendbuf, recvbuf, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, MPI_COMM_WORLD);
smpi_execute_flops(comp_size);
#ifdef HAVE_TRACING
TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
#endif
-
log_timed_action (action, clock);
}
#ifdef HAVE_TRACING
TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
#endif
-
log_timed_action (action, clock);
}
}
void *send = smpi_get_tmp_sendbuffer(send_size* smpi_datatype_size(MPI_CURRENT_TYPE));
void *recv = NULL;
-
- int root=atoi(action[4]);
- int rank = smpi_process_index();
+ int root=0;
+ if(action[4])
+ root=atoi(action[4]);
+ int rank = smpi_comm_rank(MPI_COMM_WORLD);
if(rank==root)
recv = smpi_get_tmp_recvbuffer(recv_size*comm_size* smpi_datatype_size(MPI_CURRENT_TYPE2));
extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
- TRACE_smpi_collective_in(rank, root, __FUNCTION__, extra);
+ TRACE_smpi_collective_in(smpi_process_index(), root, __FUNCTION__, extra);
#endif
mpi_coll_gather_fun(send, send_size, MPI_CURRENT_TYPE,
recv, recv_size, MPI_CURRENT_TYPE2,
root, MPI_COMM_WORLD);
#ifdef HAVE_TRACING
- TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
+ TRACE_smpi_collective_out(smpi_process_index(), -1, __FUNCTION__);
#endif
-
log_timed_action (action, clock);
}
}
int root=atoi(action[3+comm_size]);
- int rank = smpi_process_index();
+ int rank = smpi_comm_rank(MPI_COMM_WORLD);;
if(rank==root)
recv = smpi_get_tmp_recvbuffer(recv_sum* smpi_datatype_size(MPI_CURRENT_TYPE2));
extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE);
extra->datatype2 = encode_datatype(MPI_CURRENT_TYPE2);
- TRACE_smpi_collective_in(rank, root, __FUNCTION__, extra);
+ TRACE_smpi_collective_in(smpi_process_index(), root, __FUNCTION__, extra);
#endif
smpi_mpi_gatherv(send, send_size, MPI_CURRENT_TYPE,
recv, recvcounts, disps, MPI_CURRENT_TYPE2,
root, MPI_COMM_WORLD);
#ifdef HAVE_TRACING
- TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
+ TRACE_smpi_collective_out(smpi_process_index(), -1, __FUNCTION__);
#endif
log_timed_action (action, clock);
xbt_free(recvcounts);
xbt_free(disps);
-
}
static void action_reducescatter(const char *const *action) {
int *disps = xbt_new0(int, comm_size);
int i=0;
int rank = smpi_process_index();
-
+ int size = 0;
if(action[3+comm_size])
MPI_CURRENT_TYPE=decode_datatype(action[3+comm_size]);
else
for(i=0;i<comm_size;i++) {
recvcounts[i] = atoi(action[i+2]);
disps[i] = 0;
+ size+=recvcounts[i];
}
#ifdef HAVE_TRACING
extra->comp_size = comp_size;
extra->num_processes = comm_size;
-
TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
#endif
- mpi_coll_reduce_scatter_fun(NULL, NULL, recvcounts, MPI_CURRENT_TYPE, MPI_OP_NULL,
+ void *sendbuf = smpi_get_tmp_sendbuffer(size* smpi_datatype_size(MPI_CURRENT_TYPE));
+ void *recvbuf = smpi_get_tmp_recvbuffer(size* smpi_datatype_size(MPI_CURRENT_TYPE));
+
+ mpi_coll_reduce_scatter_fun(sendbuf, recvbuf, recvcounts, MPI_CURRENT_TYPE, MPI_OP_NULL,
MPI_COMM_WORLD);
smpi_execute_flops(comp_size);
void smpi_replay_init(int *argc, char***argv){
smpi_process_init(argc, argv);
smpi_process_mark_as_initialized();
+ smpi_process_set_replaying(1);
#ifdef HAVE_TRACING
int rank = smpi_process_index();
TRACE_smpi_init(rank);
active_processes--;
}
- xbt_dynar_free_container(&(reqq[smpi_process_index()]));
-
if(!active_processes){
/* Last process alive speaking */
/* end the simulated timer */
sim_time = smpi_process_simulated_elapsed();
+ }
+
+
+ xbt_dynar_free_container(&(reqq[smpi_process_index()]));
+
+ if(!active_processes){
XBT_INFO("Simulation time %f", sim_time);
_xbt_replay_action_exit();
xbt_free(sendbuffer);
xbt_free(reqq);
reqq = NULL;
}
- mpi_coll_barrier_fun(MPI_COMM_WORLD);
+
+
#ifdef HAVE_TRACING
int rank = smpi_process_index();
instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);