Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
hide this from users
[simgrid.git] / src / smpi / smpi_replay.cpp
index 93404cc..3292e7c 100644 (file)
@@ -4,13 +4,14 @@
  * under the terms of the license (GNU LGPL) which comes with this package. */
 
 #include "private.h"
-#include "xbt/replay.h"
-#include "src/smpi/smpi_group.hpp"
+#include "xbt/replay.hpp"
 #include <unordered_map>
 #include <vector>
 
 #define KEY_SIZE (sizeof(int) * 2 + 1)
 
+using namespace simgrid::smpi;
+
 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_replay,smpi,"Trace Replay with SMPI");
 
 int communicator_size = 0;
@@ -227,9 +228,9 @@ static void action_send(const char *const *action)
   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE, nullptr);
   TRACE_smpi_ptp_in(rank, rank, dst_traced, __FUNCTION__, extra);
   if (!TRACE_smpi_view_internals())
-    TRACE_smpi_send(rank, rank, dst_traced, 0, size*smpi_datatype_size(MPI_CURRENT_TYPE));
+    TRACE_smpi_send(rank, rank, dst_traced, 0, size*MPI_CURRENT_TYPE->size());
 
-  smpi_mpi_send(nullptr, size, MPI_CURRENT_TYPE, to , 0, MPI_COMM_WORLD);
+  Request::send(nullptr, size, MPI_CURRENT_TYPE, to , 0, MPI_COMM_WORLD);
 
   log_timed_action (action, clock);
 
@@ -258,12 +259,11 @@ static void action_Isend(const char *const *action)
   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE, nullptr);
   TRACE_smpi_ptp_in(rank, rank, dst_traced, __FUNCTION__, extra);
   if (!TRACE_smpi_view_internals())
-    TRACE_smpi_send(rank, rank, dst_traced, 0, size*smpi_datatype_size(MPI_CURRENT_TYPE));
+    TRACE_smpi_send(rank, rank, dst_traced, 0, size*MPI_CURRENT_TYPE->size());
 
-  MPI_Request request = smpi_mpi_isend(nullptr, size, MPI_CURRENT_TYPE, to, 0,MPI_COMM_WORLD);
+  MPI_Request request = Request::isend(nullptr, size, MPI_CURRENT_TYPE, to, 0,MPI_COMM_WORLD);
 
   TRACE_smpi_ptp_out(rank, rank, dst_traced, __FUNCTION__);
-  request->send = 1;
 
   get_reqq_self()->push_back(request);
 
@@ -295,11 +295,11 @@ static void action_recv(const char *const *action) {
 
   //unknown size from the receiver point of view
   if(size<=0.0){
-    smpi_mpi_probe(from, 0, MPI_COMM_WORLD, &status);
+    Request::probe(from, 0, MPI_COMM_WORLD, &status);
     size=status.count;
   }
 
-  smpi_mpi_recv(nullptr, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD, &status);
+  Request::recv(nullptr, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD, &status);
 
   TRACE_smpi_ptp_out(rank, src_traced, rank, __FUNCTION__);
   if (!TRACE_smpi_view_internals()) {
@@ -333,14 +333,13 @@ static void action_Irecv(const char *const *action)
   MPI_Status status;
   //unknow size from the receiver pov
   if(size<=0.0){
-      smpi_mpi_probe(from, 0, MPI_COMM_WORLD, &status);
+      Request::probe(from, 0, MPI_COMM_WORLD, &status);
       size=status.count;
   }
 
-  MPI_Request request = smpi_mpi_irecv(nullptr, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD);
+  MPI_Request request = Request::irecv(nullptr, size, MPI_CURRENT_TYPE, from, 0, MPI_COMM_WORLD);
 
   TRACE_smpi_ptp_out(rank, src_traced, rank, __FUNCTION__);
-  request->recv = 1;
   get_reqq_self()->push_back(request);
 
   log_timed_action (action, clock);
@@ -362,7 +361,7 @@ static void action_test(const char *const *action){
     extra->type=TRACING_TEST;
     TRACE_smpi_testing_in(rank, extra);
 
-    int flag = smpi_mpi_test(&request, &status);
+    int flag = Request::test(&request, &status);
 
     XBT_DEBUG("MPI_Test result: %d", flag);
     /* push back request in vector to be caught by a subsequent wait. if the test did succeed, the request is now nullptr.*/
@@ -388,17 +387,17 @@ static void action_wait(const char *const *action){
     return;
   }
 
-  int rank = request->comm != MPI_COMM_NULL ? request->comm->rank() : -1;
+  int rank = request->comm() != MPI_COMM_NULL ? request->comm()->rank() : -1;
 
-  MPI_Group group = request->comm->group();
-  int src_traced = group->rank(request->src);
-  int dst_traced = group->rank(request->dst);
-  int is_wait_for_receive = request->recv;
+  MPI_Group group = request->comm()->group();
+  int src_traced = group->rank(request->src());
+  int dst_traced = group->rank(request->dst());
+  int is_wait_for_receive = (request->flags() & RECV);
   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
   extra->type = TRACING_WAIT;
   TRACE_smpi_ptp_in(rank, src_traced, dst_traced, __FUNCTION__, extra);
 
-  smpi_mpi_wait(&request, &status);
+  Request::wait(&request, &status);
 
   TRACE_smpi_ptp_out(rank, src_traced, dst_traced, __FUNCTION__);
   if (is_wait_for_receive)
@@ -423,14 +422,14 @@ static void action_waitall(const char *const *action){
    int recvs_rcv[count_requests];
    unsigned int i=0;
    for (auto req : *(get_reqq_self())){
-     if (req && req->recv){
-       recvs_snd[i]=req->src;
-       recvs_rcv[i]=req->dst;
+     if (req && (req->flags () & RECV)){
+       recvs_snd[i]=req->src();
+       recvs_rcv[i]=req->dst();
      }else
        recvs_snd[i]=-100;
      i++;
    }
-   smpi_mpi_waitall(count_requests, &(*get_reqq_self())[0], status);
+   Request::waitall(count_requests, &(*get_reqq_self())[0], status);
 
    for (i=0; i<count_requests;i++){
      if (recvs_snd[i]!=-100)
@@ -448,7 +447,7 @@ static void action_barrier(const char *const *action){
   extra->type = TRACING_BARRIER;
   TRACE_smpi_collective_in(rank, -1, __FUNCTION__, extra);
 
-  mpi_coll_barrier_fun(MPI_COMM_WORLD);
+  Colls::barrier(MPI_COMM_WORLD);
 
   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
   log_timed_action (action, clock);
@@ -478,9 +477,9 @@ static void action_bcast(const char *const *action)
   extra->root = root_traced;
   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE, nullptr);
   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__, extra);
-  void *sendbuf = smpi_get_tmp_sendbuffer(size* smpi_datatype_size(MPI_CURRENT_TYPE));
+  void *sendbuf = smpi_get_tmp_sendbuffer(size* MPI_CURRENT_TYPE->size());
 
-  mpi_coll_bcast_fun(sendbuf, size, MPI_CURRENT_TYPE, root, MPI_COMM_WORLD);
+  Colls::bcast(sendbuf, size, MPI_CURRENT_TYPE, root, MPI_COMM_WORLD);
 
   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
   log_timed_action (action, clock);
@@ -512,9 +511,9 @@ static void action_reduce(const char *const *action)
 
   TRACE_smpi_collective_in(rank, root_traced, __FUNCTION__,extra);
 
-  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);
+  void *recvbuf = smpi_get_tmp_sendbuffer(comm_size* MPI_CURRENT_TYPE->size());
+  void *sendbuf = smpi_get_tmp_sendbuffer(comm_size* MPI_CURRENT_TYPE->size());
+  Colls::reduce(sendbuf, recvbuf, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, root, MPI_COMM_WORLD);
   smpi_execute_flops(comp_size);
 
   TRACE_smpi_collective_out(rank, root_traced, __FUNCTION__);
@@ -540,9 +539,9 @@ static void action_allReduce(const char *const *action) {
   extra->datatype1 = encode_datatype(MPI_CURRENT_TYPE, nullptr);
   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 
-  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);
+  void *recvbuf = smpi_get_tmp_sendbuffer(comm_size* MPI_CURRENT_TYPE->size());
+  void *sendbuf = smpi_get_tmp_sendbuffer(comm_size* MPI_CURRENT_TYPE->size());
+  Colls::allreduce(sendbuf, recvbuf, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, MPI_COMM_WORLD);
   smpi_execute_flops(comp_size);
 
   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
@@ -564,8 +563,8 @@ static void action_allToAll(const char *const *action) {
   else
     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 
-  void *send = smpi_get_tmp_sendbuffer(send_size*comm_size* smpi_datatype_size(MPI_CURRENT_TYPE));
-  void *recv = smpi_get_tmp_recvbuffer(recv_size*comm_size* smpi_datatype_size(MPI_CURRENT_TYPE2));
+  void *send = smpi_get_tmp_sendbuffer(send_size*comm_size* MPI_CURRENT_TYPE->size());
+  void *recv = smpi_get_tmp_recvbuffer(recv_size*comm_size* MPI_CURRENT_TYPE2->size());
 
   int rank = smpi_process_index();
   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
@@ -577,7 +576,7 @@ static void action_allToAll(const char *const *action) {
 
   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 
-  mpi_coll_alltoall_fun(send, send_size, MPI_CURRENT_TYPE, recv, recv_size, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
+  Colls::alltoall(send, send_size, MPI_CURRENT_TYPE, recv, recv_size, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
 
   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
   log_timed_action (action, clock);
@@ -605,7 +604,7 @@ static void action_gather(const char *const *action) {
   } else {
     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
   }
-  void *send = smpi_get_tmp_sendbuffer(send_size* smpi_datatype_size(MPI_CURRENT_TYPE));
+  void *send = smpi_get_tmp_sendbuffer(send_size* MPI_CURRENT_TYPE->size());
   void *recv = nullptr;
   int root=0;
   if(action[4])
@@ -613,7 +612,7 @@ static void action_gather(const char *const *action) {
   int rank = MPI_COMM_WORLD->rank();
 
   if(rank==root)
-    recv = smpi_get_tmp_recvbuffer(recv_size*comm_size* smpi_datatype_size(MPI_CURRENT_TYPE2));
+    recv = smpi_get_tmp_recvbuffer(recv_size*comm_size* MPI_CURRENT_TYPE2->size());
 
   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
   extra->type = TRACING_GATHER;
@@ -625,7 +624,7 @@ static void action_gather(const char *const *action) {
 
   TRACE_smpi_collective_in(smpi_process_index(), root, __FUNCTION__, extra);
 
-  mpi_coll_gather_fun(send, send_size, MPI_CURRENT_TYPE, recv, recv_size, MPI_CURRENT_TYPE2, root, MPI_COMM_WORLD);
+  Colls::gather(send, send_size, MPI_CURRENT_TYPE, recv, recv_size, MPI_CURRENT_TYPE2, root, MPI_COMM_WORLD);
 
   TRACE_smpi_collective_out(smpi_process_index(), -1, __FUNCTION__);
   log_timed_action (action, clock);
@@ -656,7 +655,7 @@ static void action_gatherv(const char *const *action) {
   } else
     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 
-  void *send = smpi_get_tmp_sendbuffer(send_size* smpi_datatype_size(MPI_CURRENT_TYPE));
+  void *send = smpi_get_tmp_sendbuffer(send_size* MPI_CURRENT_TYPE->size());
   void *recv = nullptr;
   for(int i=0;i<comm_size;i++) {
     recvcounts[i] = atoi(action[i+3]);
@@ -668,7 +667,7 @@ static void action_gatherv(const char *const *action) {
   int rank = MPI_COMM_WORLD->rank();
 
   if(rank==root)
-    recv = smpi_get_tmp_recvbuffer(recv_sum* smpi_datatype_size(MPI_CURRENT_TYPE2));
+    recv = smpi_get_tmp_recvbuffer(recv_sum* MPI_CURRENT_TYPE2->size());
 
   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
   extra->type = TRACING_GATHERV;
@@ -683,7 +682,7 @@ static void action_gatherv(const char *const *action) {
 
   TRACE_smpi_collective_in(smpi_process_index(), root, __FUNCTION__, extra);
 
-  smpi_mpi_gatherv(send, send_size, MPI_CURRENT_TYPE, recv, recvcounts, disps, MPI_CURRENT_TYPE2, root, MPI_COMM_WORLD);
+  Colls::gatherv(send, send_size, MPI_CURRENT_TYPE, recv, recvcounts, disps, MPI_CURRENT_TYPE2, root, MPI_COMM_WORLD);
 
   TRACE_smpi_collective_out(smpi_process_index(), -1, __FUNCTION__);
   log_timed_action (action, clock);
@@ -726,10 +725,10 @@ static void action_reducescatter(const char *const *action) {
 
   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 
-  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));
+  void *sendbuf = smpi_get_tmp_sendbuffer(size* MPI_CURRENT_TYPE->size());
+  void *recvbuf = smpi_get_tmp_recvbuffer(size* MPI_CURRENT_TYPE->size());
 
-  mpi_coll_reduce_scatter_fun(sendbuf, recvbuf, recvcounts, MPI_CURRENT_TYPE, MPI_OP_NULL, MPI_COMM_WORLD);
+  Colls::reduce_scatter(sendbuf, recvbuf, recvcounts, MPI_CURRENT_TYPE, MPI_OP_NULL, MPI_COMM_WORLD);
   smpi_execute_flops(comp_size);
 
   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
@@ -758,8 +757,8 @@ static void action_allgather(const char *const *action) {
   } else
     MPI_CURRENT_TYPE = MPI_DEFAULT_TYPE;
 
-  void *sendbuf = smpi_get_tmp_sendbuffer(sendcount* smpi_datatype_size(MPI_CURRENT_TYPE));
-  void *recvbuf = smpi_get_tmp_recvbuffer(recvcount* smpi_datatype_size(MPI_CURRENT_TYPE2));
+  void *sendbuf = smpi_get_tmp_sendbuffer(sendcount* MPI_CURRENT_TYPE->size());
+  void *recvbuf = smpi_get_tmp_recvbuffer(recvcount* MPI_CURRENT_TYPE2->size());
 
   int rank = smpi_process_index();
   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
@@ -772,7 +771,7 @@ static void action_allgather(const char *const *action) {
 
   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 
-  mpi_coll_allgather_fun(sendbuf, sendcount, MPI_CURRENT_TYPE, recvbuf, recvcount, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
+  Colls::allgather(sendbuf, sendcount, MPI_CURRENT_TYPE, recvbuf, recvcount, MPI_CURRENT_TYPE2, MPI_COMM_WORLD);
 
   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
   log_timed_action (action, clock);
@@ -802,14 +801,14 @@ static void action_allgatherv(const char *const *action) {
   } else
     MPI_CURRENT_TYPE = MPI_DEFAULT_TYPE;
 
-  void *sendbuf = smpi_get_tmp_sendbuffer(sendcount* smpi_datatype_size(MPI_CURRENT_TYPE));
+  void *sendbuf = smpi_get_tmp_sendbuffer(sendcount* MPI_CURRENT_TYPE->size());
 
   for(int i=0;i<comm_size;i++) {
     recvcounts[i] = atoi(action[i+3]);
     recv_sum=recv_sum+recvcounts[i];
     disps[i] = 0;
   }
-  void *recvbuf = smpi_get_tmp_recvbuffer(recv_sum* smpi_datatype_size(MPI_CURRENT_TYPE2));
+  void *recvbuf = smpi_get_tmp_recvbuffer(recv_sum* MPI_CURRENT_TYPE2->size());
 
   int rank = smpi_process_index();
   instr_extra_data extra = xbt_new0(s_instr_extra_data_t,1);
@@ -824,7 +823,7 @@ static void action_allgatherv(const char *const *action) {
 
   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 
-  mpi_coll_allgatherv_fun(sendbuf, sendcount, MPI_CURRENT_TYPE, recvbuf, recvcounts, disps, MPI_CURRENT_TYPE2,
+  Colls::allgatherv(sendbuf, sendcount, MPI_CURRENT_TYPE, recvbuf, recvcounts, disps, MPI_CURRENT_TYPE2,
                           MPI_COMM_WORLD);
 
   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
@@ -860,8 +859,8 @@ static void action_allToAllv(const char *const *action) {
   else
     MPI_CURRENT_TYPE=MPI_DEFAULT_TYPE;
 
-  void *sendbuf = smpi_get_tmp_sendbuffer(send_buf_size* smpi_datatype_size(MPI_CURRENT_TYPE));
-  void *recvbuf  = smpi_get_tmp_recvbuffer(recv_buf_size* smpi_datatype_size(MPI_CURRENT_TYPE2));
+  void *sendbuf = smpi_get_tmp_sendbuffer(send_buf_size* MPI_CURRENT_TYPE->size());
+  void *recvbuf  = smpi_get_tmp_recvbuffer(recv_buf_size* MPI_CURRENT_TYPE2->size());
 
   for(int i=0;i<comm_size;i++) {
     sendcounts[i] = atoi(action[i+3]);
@@ -888,7 +887,7 @@ static void action_allToAllv(const char *const *action) {
 
   TRACE_smpi_collective_in(rank, -1, __FUNCTION__,extra);
 
-  mpi_coll_alltoallv_fun(sendbuf, sendcounts, senddisps, MPI_CURRENT_TYPE,recvbuf, recvcounts, recvdisps,
+  Colls::alltoallv(sendbuf, sendcounts, senddisps, MPI_CURRENT_TYPE,recvbuf, recvcounts, recvdisps,
                          MPI_CURRENT_TYPE, MPI_COMM_WORLD);
 
   TRACE_smpi_collective_out(rank, -1, __FUNCTION__);
@@ -910,33 +909,31 @@ void smpi_replay_run(int *argc, char***argv){
   TRACE_smpi_collective_in(rank, -1, operation, extra);
   TRACE_smpi_collective_out(rank, -1, operation);
   xbt_free(operation);
-
-  if (_xbt_replay_action_init()==0) {
-    xbt_replay_action_register("init",       action_init);
-    xbt_replay_action_register("finalize",   action_finalize);
-    xbt_replay_action_register("comm_size",  action_comm_size);
-    xbt_replay_action_register("comm_split", action_comm_split);
-    xbt_replay_action_register("comm_dup",   action_comm_dup);
-    xbt_replay_action_register("send",       action_send);
-    xbt_replay_action_register("Isend",      action_Isend);
-    xbt_replay_action_register("recv",       action_recv);
-    xbt_replay_action_register("Irecv",      action_Irecv);
-    xbt_replay_action_register("test",       action_test);
-    xbt_replay_action_register("wait",       action_wait);
-    xbt_replay_action_register("waitAll",    action_waitall);
-    xbt_replay_action_register("barrier",    action_barrier);
-    xbt_replay_action_register("bcast",      action_bcast);
-    xbt_replay_action_register("reduce",     action_reduce);
-    xbt_replay_action_register("allReduce",  action_allReduce);
-    xbt_replay_action_register("allToAll",   action_allToAll);
-    xbt_replay_action_register("allToAllV",  action_allToAllv);
-    xbt_replay_action_register("gather",  action_gather);
-    xbt_replay_action_register("gatherV",  action_gatherv);
-    xbt_replay_action_register("allGather",  action_allgather);
-    xbt_replay_action_register("allGatherV",  action_allgatherv);
-    xbt_replay_action_register("reduceScatter",  action_reducescatter);
-    xbt_replay_action_register("compute",    action_compute);
-  }
+  _xbt_replay_action_init();
+  xbt_replay_action_register("init",       action_init);
+  xbt_replay_action_register("finalize",   action_finalize);
+  xbt_replay_action_register("comm_size",  action_comm_size);
+  xbt_replay_action_register("comm_split", action_comm_split);
+  xbt_replay_action_register("comm_dup",   action_comm_dup);
+  xbt_replay_action_register("send",       action_send);
+  xbt_replay_action_register("Isend",      action_Isend);
+  xbt_replay_action_register("recv",       action_recv);
+  xbt_replay_action_register("Irecv",      action_Irecv);
+  xbt_replay_action_register("test",       action_test);
+  xbt_replay_action_register("wait",       action_wait);
+  xbt_replay_action_register("waitAll",    action_waitall);
+  xbt_replay_action_register("barrier",    action_barrier);
+  xbt_replay_action_register("bcast",      action_bcast);
+  xbt_replay_action_register("reduce",     action_reduce);
+  xbt_replay_action_register("allReduce",  action_allReduce);
+  xbt_replay_action_register("allToAll",   action_allToAll);
+  xbt_replay_action_register("allToAllV",  action_allToAllv);
+  xbt_replay_action_register("gather",  action_gather);
+  xbt_replay_action_register("gatherV",  action_gatherv);
+  xbt_replay_action_register("allGather",  action_allgather);
+  xbt_replay_action_register("allGatherV",  action_allgatherv);
+  xbt_replay_action_register("reduceScatter",  action_reducescatter);
+  xbt_replay_action_register("compute",    action_compute);
 
   //if we have a delayed start, sleep here.
   if(*argc>2){
@@ -968,7 +965,7 @@ void smpi_replay_run(int *argc, char***argv){
       requests[i] = req;
       i++;
     }
-    smpi_mpi_waitall(count_requests, requests, status);
+    Request::waitall(count_requests, requests, status);
   }
   delete get_reqq_self();
   active_processes--;