Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
add an smpi/ois factor, to account for the fact that MPI_Send and MPI_Isend internal...
[simgrid.git] / src / smpi / smpi_base.c
index 2c99064..9e48e3c 100644 (file)
@@ -59,6 +59,129 @@ static int match_send(void* a, void* b,smx_action_t ignored) {
    } else return 0;
 }
 
+
+typedef struct s_smpi_factor *smpi_factor_t;
+typedef struct s_smpi_factor {
+  long factor;
+  int nb_values;
+  double values[4];//arbitrary set to 4
+} s_smpi_factor_t;
+xbt_dynar_t smpi_os_values = NULL;
+xbt_dynar_t smpi_or_values = NULL;
+xbt_dynar_t smpi_ois_values = NULL;
+
+// Methods used to parse and store the values for timing injections in smpi
+// These are taken from surf/network.c and generalized to have more factors
+// These methods should be merged with those in surf/network.c (moved somewhere in xbt ?)
+
+static int factor_cmp(const void *pa, const void *pb)
+{
+  return (((s_smpi_factor_t*)pa)->factor > ((s_smpi_factor_t*)pb)->factor);
+}
+
+
+static xbt_dynar_t parse_factor(const char *smpi_coef_string)
+{
+  char *value = NULL;
+  unsigned int iter = 0;
+  s_smpi_factor_t fact;
+  int i=0;
+  xbt_dynar_t smpi_factor, radical_elements, radical_elements2 = NULL;
+
+  smpi_factor = xbt_dynar_new(sizeof(s_smpi_factor_t), NULL);
+  radical_elements = xbt_str_split(smpi_coef_string, ";");
+  xbt_dynar_foreach(radical_elements, iter, value) {
+    fact.nb_values=0;
+    radical_elements2 = xbt_str_split(value, ":");
+    if (xbt_dynar_length(radical_elements2) <2 || xbt_dynar_length(radical_elements2) > 5)
+      xbt_die("Malformed radical for smpi factor!");
+    for(i =0; i<xbt_dynar_length(radical_elements2);i++ ){
+        if (i==0){
+           fact.factor = atol(xbt_dynar_get_as(radical_elements2, i, char *));
+        }else{
+           fact.values[fact.nb_values] = atof(xbt_dynar_get_as(radical_elements2, i, char *));
+           fact.nb_values++;
+        }
+    }
+
+    xbt_dynar_push_as(smpi_factor, s_smpi_factor_t, fact);
+    XBT_DEBUG("smpi_factor:\t%ld : %d values, first: %f", fact.factor, fact.nb_values ,fact.values[0]);
+    xbt_dynar_free(&radical_elements2);
+  }
+  xbt_dynar_free(&radical_elements);
+  iter=0;
+  xbt_dynar_sort(smpi_factor, &factor_cmp);
+  xbt_dynar_foreach(smpi_factor, iter, fact) {
+    XBT_DEBUG("smpi_factor:\t%ld : %d values, first: %f", fact.factor, fact.nb_values ,fact.values[0]);
+  }
+  return smpi_factor;
+}
+
+static double smpi_os(double size)
+{
+  if (!smpi_os_values)
+    smpi_os_values =
+        parse_factor(sg_cfg_get_string("smpi/os"));
+
+  unsigned int iter = 0;
+  s_smpi_factor_t fact;
+  double current=0.0;
+  xbt_dynar_foreach(smpi_os_values, iter, fact) {
+    if (size <= fact.factor) {
+        XBT_DEBUG("os : %lf <= %ld return %f", size, fact.factor, current);
+      return current;
+    }else{
+      current=fact.values[0]+fact.values[1]*size;
+    }
+  }
+  XBT_DEBUG("os : %lf > %ld return %f", size, fact.factor, current);
+
+  return current;
+}
+
+static double smpi_ois(double size)
+{
+  if (!smpi_ois_values)
+    smpi_ois_values =
+        parse_factor(sg_cfg_get_string("smpi/ois"));
+
+  unsigned int iter = 0;
+  s_smpi_factor_t fact;
+  double current=0.0;
+  xbt_dynar_foreach(smpi_ois_values, iter, fact) {
+    if (size <= fact.factor) {
+        XBT_DEBUG("ois : %lf <= %ld return %f", size, fact.factor, current);
+      return current;
+    }else{
+      current=fact.values[0]+fact.values[1]*size;
+    }
+  }
+  XBT_DEBUG("ois : %lf > %ld return %f", size, fact.factor, current);
+
+  return current;
+}
+
+static double smpi_or(double size)
+{
+  if (!smpi_or_values)
+    smpi_or_values =
+        parse_factor(sg_cfg_get_string("smpi/or"));
+
+  unsigned int iter = 0;
+  s_smpi_factor_t fact;
+  double current=0.0;
+  xbt_dynar_foreach(smpi_or_values, iter, fact) {
+    if (size <= fact.factor) {
+        XBT_DEBUG("or : %lf <= %ld return %f", size, fact.factor, current);
+      return current;
+    }else
+      current=fact.values[0]+fact.values[1]*size;
+  }
+  XBT_DEBUG("or : %lf > %ld return %f", size, fact.factor, current);
+
+  return current;
+}
+
 static MPI_Request build_request(void *buf, int count,
                                  MPI_Datatype datatype, int src, int dst,
                                  int tag, MPI_Comm comm, unsigned flags)
@@ -165,6 +288,16 @@ MPI_Request smpi_mpi_send_init(void *buf, int count, MPI_Datatype datatype,
   return request;
 }
 
+MPI_Request smpi_mpi_ssend_init(void *buf, int count, MPI_Datatype datatype,
+                               int dst, int tag, MPI_Comm comm)
+{
+  MPI_Request request =
+    build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag,
+                  comm, PERSISTENT | SSEND | SEND);
+  request->refcount++;
+  return request;
+}
+
 MPI_Request smpi_mpi_recv_init(void *buf, int count, MPI_Datatype datatype,
                                int src, int tag, MPI_Comm comm)
 {
@@ -191,6 +324,13 @@ void smpi_mpi_start(MPI_Request request)
     request->real_size=request->size;
     smpi_datatype_use(request->old_type);
     request->action = simcall_comm_irecv(mailbox, request->buf, &request->real_size, &match_recv, request);
+
+    double sleeptime = smpi_or(request->size);
+    if(sleeptime!=0.0){
+        simcall_process_sleep(sleeptime);
+        XBT_DEBUG("receiving size of %zu : sleep %lf ", request->size, smpi_or(request->size));
+    }
+
   } else {
 
     int receiver = smpi_group_index(smpi_comm_group(request->comm), request->dst);
@@ -205,7 +345,7 @@ void smpi_mpi_start(MPI_Request request)
       XBT_DEBUG("Send request %p is not in the permanent receive mailbox (buf: %p)",request,request->buf);
       mailbox = smpi_process_remote_mailbox(receiver);
     }
-    if (request->size < 64*1024 ) { //(FIXME: this limit should be configurable)
+    if ( (! (request->flags & SSEND)) && (request->size < sg_cfg_get_int("smpi/send_is_detached_thres"))) {
       void *oldbuf = NULL;
       request->detached = 1;
       request->refcount++;
@@ -222,6 +362,17 @@ void smpi_mpi_start(MPI_Request request)
     request->real_size=request->size;
     smpi_datatype_use(request->old_type);
 
+    //if we are giving back the control to the user without waiting for completion, we have to inject timings
+    double sleeptime =0.0;
+    if(request->detached && !(request->flags & ISEND))
+      sleeptime = smpi_os(request->size);
+    else
+      sleeptime = smpi_ois(request->size);
+
+    if(sleeptime!=0.0){
+        simcall_process_sleep(sleeptime);
+        XBT_DEBUG("sending size of %zu : sleep %lf ", request->size, smpi_os(request->size));
+    }
     request->action =
       simcall_comm_isend(mailbox, request->size, -1.0,
                          request->buf, request->real_size,
@@ -281,12 +432,24 @@ MPI_Request smpi_mpi_isend(void *buf, int count, MPI_Datatype datatype,
 {
   MPI_Request request =
       build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag,
-                    comm, NON_PERSISTENT | SEND);
+                    comm, NON_PERSISTENT | ISEND | SEND);
 
   smpi_mpi_start(request);
   return request;
 }
 
+MPI_Request smpi_mpi_issend(void *buf, int count, MPI_Datatype datatype,
+                           int dst, int tag, MPI_Comm comm)
+{
+  MPI_Request request =
+      build_request(buf, count, datatype, smpi_comm_rank(comm), dst, tag,
+                    comm, NON_PERSISTENT | ISEND | SSEND | SEND);
+  smpi_mpi_start(request);
+  return request;
+}
+
+
+
 MPI_Request smpi_irecv_init(void *buf, int count, MPI_Datatype datatype,
                             int src, int tag, MPI_Comm comm)
 {
@@ -326,6 +489,13 @@ void smpi_mpi_send(void *buf, int count, MPI_Datatype datatype, int dst,
 
 }
 
+void smpi_mpi_ssend(void *buf, int count, MPI_Datatype datatype,
+                           int dst, int tag, MPI_Comm comm)
+{
+  MPI_Request request = smpi_mpi_issend(buf, count, datatype, dst, tag, comm);
+  smpi_mpi_wait(&request, MPI_STATUS_IGNORE);
+}
+
 void smpi_mpi_sendrecv(void *sendbuf, int sendcount, MPI_Datatype sendtype,
                        int dst, int sendtag, void *recvbuf, int recvcount,
                        MPI_Datatype recvtype, int src, int recvtag,