Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
[mc] Fix MC with the class-hierarchification of simgrid::simix::Synchro
[simgrid.git] / src / smpi / smpi_base.cpp
index d9ae5ee..eeaada7 100644 (file)
@@ -4,6 +4,8 @@
 /* This program is free software; you can redistribute it and/or modify it
  * under the terms of the license (GNU LGPL) which comes with this package. */
 
+#include <xbt/config.hpp>
+
 #include "private.h"
 #include "xbt/virtu.h"
 #include "mc/mc.h"
@@ -15,6 +17,8 @@
 #include "simgrid/sg_config.h"
 #include "colls/colls.h"
 
+#include "src/simix/SynchroComm.hpp"
+
 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(smpi_base, smpi, "Logging specific to SMPI (base)");
 
 static int match_recv(void* a, void* b, smx_synchro_t ignored) {
@@ -68,13 +72,17 @@ typedef struct s_smpi_factor_multival { // FIXME: this should be merged (dedupli
   int nb_values;
   double values[4];//arbitrary set to 4
 } s_smpi_factor_multival_t;
+
 xbt_dynar_t smpi_os_values = NULL;
 xbt_dynar_t smpi_or_values = NULL;
 xbt_dynar_t smpi_ois_values = NULL;
 
-double smpi_wtime_sleep = 0.0;
-double smpi_iprobe_sleep = 1e-4;
-double smpi_test_sleep = 1e-4;
+static simgrid::config::Flag<double> smpi_wtime_sleep(
+  "smpi/wtime", "Minimum time to inject inside a call to MPI_Wtime", 0.0);
+static simgrid::config::Flag<double> smpi_iprobe_sleep(
+  "smpi/iprobe", "Minimum time to inject inside a call to MPI_Iprobe", 1e-4);
+static simgrid::config::Flag<double> smpi_test_sleep(
+  "smpi/test", "Minimum time to inject inside a call to MPI_Test", 1e-4);
 
 static int factor_cmp(const void *pa, const void *pb)
 {
@@ -126,7 +134,7 @@ static xbt_dynar_t parse_factor(const char *smpi_coef_string)
 static double smpi_os(double size)
 {
   if (!smpi_os_values) {
-    smpi_os_values = parse_factor(sg_cfg_get_string("smpi/os"));
+    smpi_os_values = parse_factor(xbt_cfg_get_string("smpi/os"));
     smpi_register_static(smpi_os_values, xbt_dynar_free_voidp);
   }
   unsigned int iter = 0;
@@ -155,7 +163,7 @@ static double smpi_os(double size)
 static double smpi_ois(double size)
 {
   if (!smpi_ois_values) {
-    smpi_ois_values = parse_factor(sg_cfg_get_string("smpi/ois"));
+    smpi_ois_values = parse_factor(xbt_cfg_get_string("smpi/ois"));
     smpi_register_static(smpi_ois_values, xbt_dynar_free_voidp);
   }
   unsigned int iter = 0;
@@ -182,7 +190,7 @@ static double smpi_ois(double size)
 static double smpi_or(double size)
 {
   if (!smpi_or_values) {
-    smpi_or_values = parse_factor(sg_cfg_get_string("smpi/or"));
+    smpi_or_values = parse_factor(xbt_cfg_get_string("smpi/or"));
     smpi_register_static(smpi_or_values, xbt_dynar_free_voidp);
   }
   unsigned int iter = 0;
@@ -237,11 +245,11 @@ static MPI_Request build_request(void *buf, int count, MPI_Datatype datatype, in
 
   s_smpi_subtype_t *subtype = static_cast<s_smpi_subtype_t*>(datatype->substruct);
 
-  if(((flags & RECV) && (flags & ACCUMULATE)) || (datatype->has_subtype == 1)){
+  if(((flags & RECV) && (flags & ACCUMULATE)) || (datatype->sizeof_substruct != 0)){
     // This part handles the problem of non-contiguous memory
     old_buf = buf;
     buf = count==0 ? NULL : xbt_malloc(count*smpi_datatype_size(datatype));
-    if ((datatype->has_subtype == 1) && (flags & SEND)) {
+    if ((datatype->sizeof_substruct != 0) && (flags & SEND)) {
       subtype->serialize(old_buf, buf, count, datatype->substruct);
     }
   }
@@ -252,10 +260,12 @@ static MPI_Request build_request(void *buf, int count, MPI_Datatype datatype, in
   request->old_type = datatype;
 
   request->size = smpi_datatype_size(datatype) * count;
+  smpi_datatype_use(datatype);
   request->src = src;
   request->dst = dst;
   request->tag = tag;
   request->comm = comm;
+  smpi_comm_use(request->comm);
   request->action = NULL;
   request->flags = flags;
   request->detached = 0;
@@ -272,7 +282,6 @@ static MPI_Request build_request(void *buf, int count, MPI_Datatype datatype, in
   request->op = MPI_REPLACE;
   request->send = 0;
   request->recv = 0;
-  if (flags & SEND) smpi_datatype_unuse(datatype);
 
   return request;
 }
@@ -334,7 +343,7 @@ void smpi_mpi_start(MPI_Request request)
   if (request->flags & RECV) {
     print_request("New recv", request);
 
-    int async_small_thresh = sg_cfg_get_int("smpi/async_small_thresh");
+    int async_small_thresh = xbt_cfg_get_int("smpi/async-small-thresh");
 
     xbt_mutex_t mut = smpi_process_mailboxes_mutex();
     if (async_small_thresh != 0 ||request->flags & RMA)
@@ -382,8 +391,6 @@ void smpi_mpi_start(MPI_Request request)
 
     // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later
     request->real_size=request->size;
-    smpi_datatype_use(request->old_type);
-    smpi_comm_use(request->comm);
     request->action = simcall_comm_irecv(SIMIX_process_self(), mailbox, request->buf, &request->real_size, &match_recv,
                                          !smpi_process_get_replaying()? &smpi_comm_copy_buffer_callback
                                          : &smpi_comm_null_copy_buffer_callback, request, -1.0);
@@ -412,7 +419,7 @@ void smpi_mpi_start(MPI_Request request)
         XBT_DEBUG("sending size of %zu : sleep %f ", request->size, smpi_os(request->size));
     }
 
-    int async_small_thresh = sg_cfg_get_int("smpi/async_small_thresh");
+    int async_small_thresh = xbt_cfg_get_int("smpi/async-small-thresh");
 
     xbt_mutex_t mut=smpi_process_remote_mailboxes_mutex(receiver);
 
@@ -448,13 +455,13 @@ void smpi_mpi_start(MPI_Request request)
     }
 
     void* buf = request->buf;
-    if ( (! (request->flags & SSEND)) &&
-         (static_cast<int>(request->size) < sg_cfg_get_int("smpi/send_is_detached_thresh"))) {
+    if ( (! (request->flags & SSEND)) && ((request->flags & RMA) ||
+         (static_cast<int>(request->size) < xbt_cfg_get_int("smpi/send-is-detached-thresh")))) {
       void *oldbuf = NULL;
       request->detached = 1;
       XBT_DEBUG("Send request %p is detached", request);
       request->refcount++;
-      if(request->old_type->has_subtype == 0){
+      if(request->old_type->sizeof_substruct == 0){
         oldbuf = request->buf;
         if (!smpi_process_get_replaying() && oldbuf && request->size!=0){
           if((smpi_privatize_global_variables)
@@ -472,8 +479,6 @@ void smpi_mpi_start(MPI_Request request)
 
     // we make a copy here, as the size is modified by simix, and we may reuse the request in another receive later
     request->real_size=request->size;
-    smpi_datatype_use(request->old_type);
-    smpi_comm_use(request->comm);
     request->action = simcall_comm_isend(SIMIX_process_from_PID(request->src+1), mailbox, request->size, -1.0,
                                          buf, request->real_size, &match_send,
                          &xbt_free_f, // how to free the userdata if a detached send fails
@@ -508,6 +513,8 @@ void smpi_mpi_request_free(MPI_Request * request)
     if((*request)->refcount<0) xbt_die("wrong refcount");
 
     if((*request)->refcount==0){
+        smpi_datatype_unuse((*request)->old_type);
+        smpi_comm_unuse((*request)->comm);
         print_request("Destroying", (*request));
         xbt_free(*request);
         *request = MPI_REQUEST_NULL;
@@ -530,6 +537,7 @@ MPI_Request smpi_rma_send_init(void *buf, int count, MPI_Datatype datatype, int
     request = build_request(buf==MPI_BOTTOM ? (void*)0 : buf, count, datatype,  src, dst, tag,
                             comm, RMA | NON_PERSISTENT | ISEND | SEND | PREPARED | ACCUMULATE);
     request->op = op;
+//    smpi_datatype_use(datatype);
   }
   return request;
 }
@@ -545,6 +553,7 @@ MPI_Request smpi_rma_recv_init(void *buf, int count, MPI_Datatype datatype, int
     request = build_request(buf==MPI_BOTTOM ? (void*)0 : buf, count, datatype,  src, dst, tag,
                             comm, RMA | NON_PERSISTENT | RECV | PREPARED | ACCUMULATE);
     request->op = op;
+//    smpi_datatype_use(datatype);
   }
   return request;
 }
@@ -671,7 +680,7 @@ static void finish_wait(MPI_Request * request, MPI_Status * status)
     print_request("Finishing", req);
     MPI_Datatype datatype = req->old_type;
 
-    if((req->flags & ACCUMULATE) || (datatype->has_subtype == 1)){
+    if((req->flags & ACCUMULATE) || (datatype->sizeof_substruct != 0)){
       if (!smpi_process_get_replaying()){
         if( smpi_privatize_global_variables && ((char*)req->old_buf >= smpi_start_data_exe)
             && ((char*)req->old_buf < smpi_start_data_exe + smpi_size_data_exe )){
@@ -680,20 +689,19 @@ static void finish_wait(MPI_Request * request, MPI_Status * status)
         }
       }
 
-      if(datatype->has_subtype == 1){
+      if(datatype->sizeof_substruct != 0){
         // This part handles the problem of non-contignous memory the unserialization at the reception
         s_smpi_subtype_t *subtype = static_cast<s_smpi_subtype_t*>(datatype->substruct);
         if(req->flags & RECV)
           subtype->unserialize(req->buf, req->old_buf, req->real_size/smpi_datatype_size(datatype) ,
                                datatype->substruct, req->op);
-        if(req->detached == 0) free(req->buf);
+        xbt_free(req->buf);
       }else if(req->flags & RECV){//apply op on contiguous buffer for accumulate
           int n =req->real_size/smpi_datatype_size(datatype);
           smpi_op_apply(req->op, req->buf, req->old_buf, &n, &datatype);
+          xbt_free(req->buf);
       }
     }
-    smpi_comm_unuse(req->comm);
-    smpi_datatype_unuse(datatype);
   }
 
   if (TRACE_smpi_view_internals()) {
@@ -836,7 +844,7 @@ void smpi_mpi_iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status*
 
   print_request("New iprobe", request);
   // We have to test both mailboxes as we don't know if we will receive one one or another
-  if (sg_cfg_get_int("smpi/async_small_thresh")>0){
+  if (xbt_cfg_get_int("smpi/async-small-thresh")>0){
       mailbox = smpi_process_mailbox_small();
       XBT_DEBUG("trying to probe the perm recv mailbox");
       request->action = simcall_comm_iprobe(mailbox, 0, request->src, request->tag, &match_recv, (void*)request);
@@ -848,7 +856,8 @@ void smpi_mpi_iprobe(int source, int tag, MPI_Comm comm, int* flag, MPI_Status*
   }
 
   if (request->action){
-    MPI_Request req = (MPI_Request)SIMIX_comm_get_src_data(request->action);
+    simgrid::simix::Comm *sync_comm = static_cast<simgrid::simix::Comm*>(request->action);
+    MPI_Request req = (MPI_Request)sync_comm->src_data;
     *flag = 1;
     if(status != MPI_STATUS_IGNORE && !(req->flags & PREPARED)) {
       status->MPI_SOURCE = smpi_group_rank(smpi_comm_group(comm), req->src);
@@ -875,14 +884,10 @@ void smpi_mpi_wait(MPI_Request * request, MPI_Status * status)
     return;
   }
 
-  if ((*request)->action != NULL) { // this is not a detached send
+  if ((*request)->action != NULL)
+    // this is not a detached send
     simcall_comm_wait((*request)->action, -1.0);
 
-  if((MC_is_active() || MC_record_replay_is_active()) && (*request)->action)
-    (*request)->action->comm.dst_data = NULL; // dangling pointer : dst_data is freed with a wait, need to set it to
-                                              // NULL for system state comparison
-  }
-
   finish_wait(request, status);
   if (*request != MPI_REQUEST_NULL && ((*request)->flags & NON_PERSISTENT))
       *request = MPI_REQUEST_NULL;
@@ -1085,7 +1090,7 @@ void smpi_mpi_reduce_scatter(void *sendbuf, void *recvbuf, int *recvcounts, MPI_
 {
     int i, size, count;
     int *displs;
-    int rank = smpi_process_index();
+    int rank = smpi_comm_rank(comm);
     void *tmpbuf;
 
     /* arbitrarily choose root as rank 0 */
@@ -1306,10 +1311,7 @@ void smpi_mpi_reduce(void *sendbuf, void *recvbuf, int count, MPI_Datatype datat
   void **tmpbufs;
 
   char* sendtmpbuf = (char*) sendbuf;
-  if( sendbuf == MPI_IN_PLACE ) {
-    sendtmpbuf = (char *)smpi_get_tmp_sendbuffer(count*smpi_datatype_get_extent(datatype));
-    smpi_datatype_copy(recvbuf, count, datatype,sendtmpbuf, count, datatype);
-  }
+
 
   rank = smpi_comm_rank(comm);
   size = smpi_comm_size(comm);
@@ -1318,6 +1320,11 @@ void smpi_mpi_reduce(void *sendbuf, void *recvbuf, int count, MPI_Datatype datat
     smpi_coll_tuned_reduce_ompi_basic_linear(sendtmpbuf, recvbuf, count, datatype, op, root, comm);
     return;
   }
+
+  if( sendbuf == MPI_IN_PLACE ) {
+    sendtmpbuf = (char *)smpi_get_tmp_sendbuffer(count*smpi_datatype_get_extent(datatype));
+    smpi_datatype_copy(recvbuf, count, datatype,sendtmpbuf, count, datatype);
+  }
   
   if(rank != root) {
     // Send buffer to root
@@ -1365,9 +1372,9 @@ void smpi_mpi_reduce(void *sendbuf, void *recvbuf, int count, MPI_Datatype datat
     xbt_free(tmpbufs);
     xbt_free(requests);
 
-    if( sendbuf == MPI_IN_PLACE ) {
-      smpi_free_tmp_buffer(sendtmpbuf);
-    }
+  }
+  if( sendbuf == MPI_IN_PLACE ) {
+    smpi_free_tmp_buffer(sendtmpbuf);
   }
 }