Logo AND Algorithmique Numérique Distribuée
Public GIT Repository
projects / simgrid.git / blobdiff
? search:
summary | shortlog | log | commit | commitdiff | tree
raw | inline | side by side
one step further on C++ization of replay
[simgrid.git] / src / smpi / smpi_replay.cpp
diff --git a/src/smpi/smpi_replay.cpp b/src/smpi/smpi_replay.cpp
index 6d7d878..282eb70 100644 (file)
--- a/src/smpi/smpi_replay.cpp
+++ b/src/smpi/smpi_replay.cpp
@@ -4,7 +4,7 @@
  * under the terms of the license (GNU LGPL) which comes with this package. */
 
 #include "private.h"
-#include "xbt/replay.h"
+#include "xbt/replay.hpp"
 #include <unordered_map>
 #include <vector>
 
@@ -226,9 +226,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);
 
@@ -257,12 +257,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);
 
@@ -294,11 +293,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()) {
@@ -332,14 +331,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);
@@ -361,7 +359,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.*/
@@ -387,17 +385,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)
@@ -422,14 +420,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)
@@ -477,7 +475,7 @@ 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);
 
@@ -511,8 +509,8 @@ 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));
+  void *recvbuf = smpi_get_tmp_sendbuffer(comm_size* MPI_CURRENT_TYPE->size());
+  void *sendbuf = smpi_get_tmp_sendbuffer(comm_size* MPI_CURRENT_TYPE->size());
   mpi_coll_reduce_fun(sendbuf, recvbuf, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, root, MPI_COMM_WORLD);
   smpi_execute_flops(comp_size);
 
@@ -539,8 +537,8 @@ 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));
+  void *recvbuf = smpi_get_tmp_sendbuffer(comm_size* MPI_CURRENT_TYPE->size());
+  void *sendbuf = smpi_get_tmp_sendbuffer(comm_size* MPI_CURRENT_TYPE->size());
   mpi_coll_allreduce_fun(sendbuf, recvbuf, comm_size, MPI_CURRENT_TYPE, MPI_OP_NULL, MPI_COMM_WORLD);
   smpi_execute_flops(comp_size);
 
@@ -563,8 +561,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);
@@ -604,7 +602,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])
@@ -612,7 +610,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;
@@ -655,7 +653,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]);
@@ -667,7 +665,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;
@@ -725,8 +723,8 @@ 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);
   smpi_execute_flops(comp_size);
@@ -757,8 +755,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);
@@ -801,14 +799,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);
@@ -859,8 +857,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]);
@@ -909,33 +907,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){
@@ -967,7 +963,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--;
Clone of SimGrid repository with local patches