X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/0a14591b7a8a433bff7aeca6b7b430b7ae27336b..eccc07ce42fe46f17c82bb0f950a818d8f603835:/src/smpi/smpi_global.c

diff --git a/src/smpi/smpi_global.c b/src/smpi/smpi_global.c
index 84d0a7ec10..4f001b6736 100644
--- a/src/smpi/smpi_global.c
+++ b/src/smpi/smpi_global.c
@@ -28,16 +28,23 @@ typedef struct s_smpi_process_data {
   smx_rdv_t mailbox_small;
   xbt_os_timer_t timer;
   MPI_Comm comm_self;
+  MPI_Comm comm_intra;
+  MPI_Comm* comm_world;
   void *data;                   /* user data */
   int index;
   char state;
   int sampling;                 /* inside an SMPI_SAMPLE_ block? */
+  char* instance_id;
+  int replaying;                /* is the process replaying a trace */
+  xbt_bar_t finalization_barrier;
 } s_smpi_process_data_t;
 
 static smpi_process_data_t *process_data = NULL;
-static int process_count = 0;
-
-MPI_Comm MPI_COMM_WORLD = MPI_COMM_NULL;
+int process_count = 0;
+int* index_to_process_data = NULL;
+extern double smpi_total_benched_time;
+xbt_os_timer_t global_timer;
+MPI_Comm MPI_COMM_WORLD = MPI_COMM_UNINITIALIZED;
 int MPI_UNIVERSE_SIZE;
 
 MPI_Errhandler *MPI_ERRORS_RETURN = NULL;
@@ -62,33 +69,48 @@ static char *get_mailbox_name_small(char *str, int index)
 
 void smpi_process_init(int *argc, char ***argv)
 {
-  int index;
+  int index=-1;
   smpi_process_data_t data;
   smx_process_t proc;
 
   if (argc && argv) {
     proc = SIMIX_process_self();
-    index = atoi((*argv)[1]);
-#ifdef SMPI_F2C
-    smpi_current_rank = index;
-#endif
-
+    //FIXME: dirty cleanup method to avoid using msg cleanup functions on these processes when using MSG+SMPI
+    proc->context->cleanup_func=SIMIX_process_cleanup;
+    char* instance_id = (*argv)[1];
+    int rank = atoi((*argv)[2]);
+    index = smpi_process_index_of_smx_process(proc);
+
+    if(!index_to_process_data){
+        index_to_process_data=(int*)xbt_malloc(SIMIX_process_count()*sizeof(int));
+    }
+    MPI_Comm* temp_comm_world;
+    xbt_bar_t temp_bar;
+    smpi_deployment_register_process(instance_id, rank, index, &temp_comm_world ,&temp_bar);
     data = smpi_process_remote_data(index);
+    data->comm_world = temp_comm_world;
+    if(temp_bar != NULL) data->finalization_barrier = temp_bar;
+    data->index = index;
+    data->instance_id = instance_id;
+    data->replaying = 0;
+    xbt_free(simcall_process_get_data(proc));
     simcall_process_set_data(proc, data);
-    if (*argc > 2) {
+    if (*argc > 3) {
       free((*argv)[1]);
-      memmove(&(*argv)[1], &(*argv)[2], sizeof(char *) * (*argc - 2));
+      memmove(&(*argv)[0], &(*argv)[2], sizeof(char *) * (*argc - 2));
       (*argv)[(*argc) - 1] = NULL;
+      (*argv)[(*argc) - 2] = NULL;
     }
-    (*argc)--;
+    (*argc)-=2;
     data->argc = argc;
     data->argv = argv;
     // set the process attached to the mailbox
     simcall_rdv_set_receiver(data->mailbox_small, proc);
+
     XBT_DEBUG("<%d> New process in the game: %p", index, proc);
 
     if(smpi_privatize_global_variables){
-      switch_data_segment(index);
+      smpi_switch_data_segment(index);
     }
 
   }
@@ -100,9 +122,9 @@ void smpi_process_destroy(void)
 {
   int index = smpi_process_index();
   if(smpi_privatize_global_variables){
-    switch_data_segment(index);
+    smpi_switch_data_segment(index);
   }
-  process_data[index]->state = SMPI_FINALIZED;
+  process_data[index_to_process_data[index]]->state = SMPI_FINALIZED;
   XBT_DEBUG("<%d> Process left the game", index);
 }
 
@@ -111,44 +133,14 @@ void smpi_process_destroy(void)
  */
 void smpi_process_finalize(void)
 {
-#if 0
-  // wait for all pending asynchronous comms to finish
-  while (SIMIX_process_has_pending_comms(SIMIX_process_self())) {
-    simcall_process_sleep(0.01);
-  }
-#else
-  int i;
-  int size = smpi_comm_size(MPI_COMM_WORLD);
-  int rank = smpi_comm_rank(MPI_COMM_WORLD);
-  /* All non-root send & receive zero-length message. */
-  if (rank > 0) {
-    smpi_mpi_ssend (NULL, 0, MPI_BYTE, 0,
-                    COLL_TAG_BARRIER,
-                    MPI_COMM_WORLD);
-    smpi_mpi_recv (NULL, 0, MPI_BYTE, 0,
-                    COLL_TAG_BARRIER,
-                    MPI_COMM_WORLD, MPI_STATUS_IGNORE);
-  }
-  /* The root collects and broadcasts the messages. */
-  else {
-    MPI_Request* requests;
-    requests = (MPI_Request*)malloc( size * sizeof(MPI_Request) );
-    for (i = 1; i < size; ++i) {
-      requests[i] = smpi_mpi_irecv(NULL, 0, MPI_BYTE, MPI_ANY_SOURCE,
-                                   COLL_TAG_BARRIER, MPI_COMM_WORLD
-                                   );
-    }
-    smpi_mpi_waitall( size-1, requests+1, MPI_STATUSES_IGNORE );
-    for (i = 1; i < size; ++i) {
-      requests[i] = smpi_mpi_issend(NULL, 0, MPI_BYTE, i,
-                                   COLL_TAG_BARRIER,
-                                   MPI_COMM_WORLD
-                                   );
-    }
-    smpi_mpi_waitall( size-1, requests+1, MPI_STATUSES_IGNORE );
-    free( requests );
-  }
-#endif
+    // This leads to an explosion of the search graph
+    // which cannot be reduced:
+    if(MC_is_active())
+      return;
+
+    int index = smpi_process_index();
+    // wait for all pending asynchronous comms to finish
+    xbt_barrier_wait(process_data[index_to_process_data[index]]->finalization_barrier);
 }
 
 /**
@@ -158,7 +150,7 @@ int smpi_process_finalized()
 {
   int index = smpi_process_index();
     if (index != MPI_UNDEFINED)
-      return (process_data[index]->state == SMPI_FINALIZED);
+      return (process_data[index_to_process_data[index]]->state == SMPI_FINALIZED);
     else
       return 0;
 }
@@ -170,7 +162,7 @@ int smpi_process_initialized(void)
 {
   int index = smpi_process_index();
   return ( (index != MPI_UNDEFINED)
-          && (process_data[index]->state == SMPI_INITIALIZED));
+          && (process_data[index_to_process_data[index]]->state == SMPI_INITIALIZED));
 }
 
 /**
@@ -179,37 +171,23 @@ int smpi_process_initialized(void)
 void smpi_process_mark_as_initialized(void)
 {
   int index = smpi_process_index();
-  if ((index != MPI_UNDEFINED) && (!process_data[index]->state != SMPI_FINALIZED))
-    process_data[index]->state = SMPI_INITIALIZED;
+  if ((index != MPI_UNDEFINED) && (process_data[index_to_process_data[index]]->state != SMPI_FINALIZED))
+    process_data[index_to_process_data[index]]->state = SMPI_INITIALIZED;
 }
 
-
-#ifdef SMPI_F2C
-int smpi_process_argc(void)
-{
-  smpi_process_data_t data = smpi_process_data();
-  return data->argc ? *(data->argc) - 1 : 0;
+void smpi_process_set_replaying(int value){
+  int index = smpi_process_index();
+  if ((index != MPI_UNDEFINED) && (process_data[index_to_process_data[index]]->state != SMPI_FINALIZED))
+    process_data[index_to_process_data[index]]->replaying = value;
 }
 
-int smpi_process_getarg(integer * index, char *dst, ftnlen len)
-{
-  smpi_process_data_t data = smpi_process_data();
-  char *arg;
-  ftnlen i;
-
-  if (!data->argc || !data->argv || *index < 1 || *index >= *(data->argc)) {
-    return -1;
-  }
-  arg = (*data->argv)[*index];
-  for (i = 0; i < len && arg[i] != '\0'; i++) {
-    dst[i] = arg[i];
-  }
-  for (; i < len; i++) {
-    dst[i] = ' ';
-  }
-  return 0;
+int smpi_process_get_replaying(){
+  int index = smpi_process_index();
+  if (index != MPI_UNDEFINED)
+    return process_data[index_to_process_data[index]]->replaying;
+  else return _xbt_replay_is_active();
 }
-#endif
+
 
 int smpi_global_size(void)
 {
@@ -230,7 +208,7 @@ smpi_process_data_t smpi_process_data(void)
 
 smpi_process_data_t smpi_process_remote_data(int index)
 {
-  return process_data[index];
+  return process_data[index_to_process_data[index]];
 }
 
 void smpi_process_set_user_data(void *data)
@@ -257,6 +235,13 @@ int smpi_process_index(void)
   return data ? data->index : MPI_UNDEFINED;
 }
 
+MPI_Comm smpi_process_comm_world(void)
+{
+  smpi_process_data_t data = smpi_process_data();
+  //return MPI_COMM_NULL if not initialized
+  return data ? *data->comm_world : MPI_COMM_NULL;
+}
+
 smx_rdv_t smpi_process_mailbox(void)
 {
   smpi_process_data_t data = smpi_process_data();
@@ -312,6 +297,18 @@ MPI_Comm smpi_process_comm_self(void)
   return data->comm_self;
 }
 
+MPI_Comm smpi_process_get_comm_intra(void)
+{
+  smpi_process_data_t data = smpi_process_data();
+  return data->comm_intra;
+}
+
+void smpi_process_set_comm_intra(MPI_Comm comm)
+{
+  smpi_process_data_t data = smpi_process_data();
+  data->comm_intra = comm;
+}
+
 void smpi_process_set_sampling(int s)
 {
   smpi_process_data_t data = smpi_process_data();
@@ -324,19 +321,19 @@ int smpi_process_get_sampling(void)
   return data->sampling;
 }
 
+
 void print_request(const char *message, MPI_Request request)
 {
-  XBT_DEBUG
+  XBT_VERB
       ("%s  request %p  [buf = %p, size = %zu, src = %d, dst = %d, tag = %d, flags = %x]",
        message, request, request->buf, request->size, request->src,
        request->dst, request->tag, request->flags);
 }
 
-static void smpi_comm_copy_buffer_callback(smx_action_t comm,
+void smpi_comm_copy_buffer_callback(smx_synchro_t comm,
                                            void *buff, size_t buff_size)
 {
   XBT_DEBUG("Copy the data over");
-  if(_xbt_replay_is_active()) return;
   void* tmpbuff=buff;
 
   if((smpi_privatize_global_variables)
@@ -344,7 +341,7 @@ static void smpi_comm_copy_buffer_callback(smx_action_t comm,
       && ((char*)buff < start_data_exe + size_data_exe )
     ){
        XBT_DEBUG("Privatization : We are copying from a zone inside global memory... Saving data to temp buffer !");
-       switch_data_segment(((smpi_process_data_t)SIMIX_process_get_data(comm->comm.src_proc))->index);
+       smpi_switch_data_segment(((smpi_process_data_t)SIMIX_process_get_data(comm->comm.src_proc))->index);
        tmpbuff = (void*)xbt_malloc(buff_size);
        memcpy(tmpbuff, buff, buff_size);
   }
@@ -355,7 +352,7 @@ static void smpi_comm_copy_buffer_callback(smx_action_t comm,
       && ((char*)comm->comm.dst_buff < start_data_exe + size_data_exe )
     ){
        XBT_DEBUG("Privatization : We are copying to a zone inside global memory - Switch data segment");
-       switch_data_segment(((smpi_process_data_t)SIMIX_process_get_data(comm->comm.dst_proc))->index);
+       smpi_switch_data_segment(((smpi_process_data_t)SIMIX_process_get_data(comm->comm.dst_proc))->index);
   }
 
 
@@ -375,6 +372,13 @@ static void smpi_comm_copy_buffer_callback(smx_action_t comm,
 
 }
 
+
+void smpi_comm_null_copy_buffer_callback(smx_synchro_t comm,
+                                           void *buff, size_t buff_size)
+{
+  return;
+}
+
 static void smpi_check_options(){
   //check correctness of MPI parameters
 
@@ -385,22 +389,33 @@ static void smpi_check_options(){
      XBT_INFO("You did not set the power of the host running the simulation.  "
               "The timings will certainly not be accurate.  "
               "Use the option \"--cfg=smpi/running_power:<flops>\" to set its value."
-              "Check http://simgrid.org/simgrid/latest/doc/options.html#options_smpi_bench for more information. ");
+              "Check http://simgrid.org/simgrid/latest/doc/options.html#options_smpi_bench for more information.");
    }
 }
 
+int smpi_enabled(void) {
+  return process_data != NULL;
+}
+
 void smpi_global_init(void)
 {
   int i;
   MPI_Group group;
   char name[MAILBOX_NAME_MAXLEN];
+  int smpirun=0;
 
-  SIMIX_comm_set_copy_data_callback(&smpi_comm_copy_buffer_callback);
-  process_count = SIMIX_process_count();
-  process_data = xbt_new(smpi_process_data_t, process_count);
+  if (!MC_is_active()) {
+    global_timer = xbt_os_timer_new();
+    xbt_os_walltimer_start(global_timer);
+  }
+  if (process_count == 0){
+    process_count = SIMIX_process_count();
+    smpirun=1;
+  }
+  process_data = xbt_new0(smpi_process_data_t, process_count);
   for (i = 0; i < process_count; i++) {
     process_data[i] = xbt_new(s_smpi_process_data_t, 1);
-    process_data[i]->index = i;
+    //process_data[i]->index = i;
     process_data[i]->argc = NULL;
     process_data[i]->argv = NULL;
     process_data[i]->mailbox = simcall_rdv_create(get_mailbox_name(name, i));
@@ -410,29 +425,27 @@ void smpi_global_init(void)
     if (MC_is_active())
       MC_ignore_heap(process_data[i]->timer, xbt_os_timer_size());
     process_data[i]->comm_self = MPI_COMM_NULL;
+    process_data[i]->comm_intra = MPI_COMM_NULL;
+    process_data[i]->comm_world = NULL;
     process_data[i]->state = SMPI_UNINITIALIZED;
     process_data[i]->sampling = 0;
+    process_data[i]->finalization_barrier = NULL;
   }
-  group = smpi_group_new(process_count);
-  MPI_COMM_WORLD = smpi_comm_new(group, NULL);
-  MPI_UNIVERSE_SIZE = smpi_comm_size(MPI_COMM_WORLD);
-  for (i = 0; i < process_count; i++) {
-    smpi_group_set_mapping(group, i, i);
-  }
-
-  //check correctness of MPI parameters
-
-  xbt_assert(sg_cfg_get_int("smpi/async_small_thres") <=
-             sg_cfg_get_int("smpi/send_is_detached_thres"));
-
-  if (sg_cfg_is_default_value("smpi/running_power")) {
-    XBT_INFO("You did not set the power of the host running the simulation.  "
-             "The timings will certainly not be accurate.  "
-             "Use the option \"--cfg=smpi/running_power:<flops>\" to set its value."
-             "Check http://simgrid.org/simgrid/latest/doc/options.html#options_smpi_bench for more information. ");
+  //if the process was launched through smpirun script
+  //we generate a global mpi_comm_world
+  //if not, we let MPI_COMM_NULL, and the comm world
+  //will be private to each mpi instance
+  if(smpirun){
+    group = smpi_group_new(process_count);
+    MPI_COMM_WORLD = smpi_comm_new(group, NULL);
+    xbt_bar_t bar=xbt_barrier_init(process_count);
+
+    MPI_UNIVERSE_SIZE = smpi_comm_size(MPI_COMM_WORLD);
+    for (i = 0; i < process_count; i++) {
+      smpi_group_set_mapping(group, i, i);
+      process_data[i]->finalization_barrier = bar;
+    }
   }
-  if(smpi_privatize_global_variables)
-    smpi_initialize_global_memory_segments();
 }
 
 void smpi_global_destroy(void)
@@ -441,14 +454,23 @@ void smpi_global_destroy(void)
   int i;
 
   smpi_bench_destroy();
-  while (smpi_group_unuse(smpi_comm_group(MPI_COMM_WORLD)) > 0);
-  xbt_free(MPI_COMM_WORLD);
+  if (MPI_COMM_WORLD != MPI_COMM_UNINITIALIZED){
+      while (smpi_group_unuse(smpi_comm_group(MPI_COMM_WORLD)) > 0);
+      xbt_free(MPI_COMM_WORLD);
+      xbt_barrier_destroy(process_data[0]->finalization_barrier);
+  }else{
+      smpi_deployment_cleanup_instances();
+  }
   MPI_COMM_WORLD = MPI_COMM_NULL;
   for (i = 0; i < count; i++) {
     if(process_data[i]->comm_self!=MPI_COMM_NULL){
       smpi_group_unuse(smpi_comm_group(process_data[i]->comm_self));
       smpi_comm_destroy(process_data[i]->comm_self);
     }
+    if(process_data[i]->comm_intra!=MPI_COMM_NULL){
+      smpi_group_unuse(smpi_comm_group(process_data[i]->comm_intra));
+      smpi_comm_destroy(process_data[i]->comm_intra);
+    }
     xbt_os_timer_free(process_data[i]->timer);
     simcall_rdv_destroy(process_data[i]->mailbox);
     simcall_rdv_destroy(process_data[i]->mailbox_small);
@@ -456,17 +478,13 @@ void smpi_global_destroy(void)
   }
   xbt_free(process_data);
   process_data = NULL;
+
+  xbt_free(index_to_process_data);
   if(smpi_privatize_global_variables)
     smpi_destroy_global_memory_segments();
   smpi_free_static();
 }
 
-/* Fortran specific stuff */
-/* With smpicc, the following weak symbols are used */
-/* With smpiff, the following weak symbols are replaced by those in libf2c */
-int __attribute__ ((weak)) xargc;
-char ** __attribute__ ((weak)) xargv;
-
 #ifndef WIN32
 void __attribute__ ((weak)) user_main_()
 {
@@ -486,10 +504,6 @@ int __attribute__ ((weak)) main(int argc, char **argv)
   return smpi_main(smpi_simulated_main_, argc, argv);
 }
 
-int __attribute__ ((weak)) MAIN__()
-{
-  return smpi_main(smpi_simulated_main_, xargc, xargv);
-};
 #endif
 
 static void smpi_init_logs(){
@@ -514,6 +528,7 @@ static void smpi_init_logs(){
   XBT_LOG_CONNECT(smpi_mpi_dt);
   XBT_LOG_CONNECT(smpi_pmpi);
   XBT_LOG_CONNECT(smpi_replay);
+  XBT_LOG_CONNECT(smpi_rma);
 
 }
 
@@ -620,14 +635,14 @@ int smpi_main(int (*realmain) (int argc, char *argv[]), int argc, char *argv[])
 
   SIMIX_global_init(&argc, argv);
 
+  smpi_init_options();
+
   // parse the platform file: get the host list
   SIMIX_create_environment(argv[1]);
   SIMIX_comm_set_copy_data_callback(&smpi_comm_copy_buffer_callback);
   SIMIX_function_register_default(realmain);
   SIMIX_launch_application(argv[2]);
 
-  smpi_init_options();
-
   smpi_global_init();
 
   smpi_check_options();
@@ -639,13 +654,25 @@ int smpi_main(int (*realmain) (int argc, char *argv[]), int argc, char *argv[])
   fflush(stdout);
   fflush(stderr);
 
-  if (MC_is_active())
+  if (MC_is_active()) {
     MC_do_the_modelcheck_for_real();
-  else
+  } else {
+  
     SIMIX_run();
 
-  if (sg_cfg_get_boolean("smpi/display_timing"))
-    XBT_INFO("Simulation time: %g seconds.", SIMIX_get_clock());
+    xbt_os_walltimer_stop(global_timer);
+    if (sg_cfg_get_boolean("smpi/display_timing")){
+      double global_time = xbt_os_timer_elapsed(global_timer);
+      XBT_INFO("Simulated time: %g seconds. \n "
+          "The simulation took %g seconds (after parsing and platform setup)\n"
+          "%g seconds were actual computation of the application"
+          , SIMIX_get_clock(), global_time , smpi_total_benched_time);
+          
+      if (smpi_total_benched_time/global_time>=0.75)
+      XBT_INFO("More than 75%% of the time was spent inside the application code.\n"
+      "You may want to use sampling functions or trace replay to reduce this.");
+    }
+  }
 
   smpi_global_destroy();
 
@@ -655,3 +682,23 @@ int smpi_main(int (*realmain) (int argc, char *argv[]), int argc, char *argv[])
 
   return 0;
 }
+
+// This function can be called from extern file, to initialize logs, options, and processes of smpi
+// without the need of smpirun
+void SMPI_init(){
+  smpi_init_logs();
+  smpi_init_options();
+  smpi_global_init();
+  smpi_check_options();
+#ifdef HAVE_TRACING
+  if (TRACE_is_enabled() && TRACE_is_configured()) {
+    TRACE_smpi_alloc();
+  }
+#endif
+  if(smpi_privatize_global_variables)
+    smpi_initialize_global_memory_segments();
+}
+
+void SMPI_finalize(){
+  smpi_global_destroy();
+}