Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Rename smx_process to ActorImpl
[simgrid.git] / src / simix / smx_global.cpp
index dbfb7f9..c2229a7 100644 (file)
@@ -4,10 +4,17 @@
 /* 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 <functional>
+#include <memory>
+
 #include <signal.h> /* Signal handling */
 #include <stdlib.h>
 #include "src/internal_config.h"
 
+#include <xbt/functional.hpp>
+
+#include <simgrid/s4u/host.hpp>
+
 #include "src/surf/surf_interface.hpp"
 #include "src/surf/storage_interface.hpp"
 #include "src/surf/xml/platf.hpp"
 #include "src/mc/mc_replay.h"
 #include "simgrid/sg_config.h"
 
-#include "src/simix/SynchroExec.hpp"
-#include "src/simix/SynchroComm.hpp"
-#include "src/simix/SynchroSleep.hpp"
-#include "src/simix/SynchroIo.hpp"
-#include "src/simix/SynchroRaw.hpp"
+#include "src/kernel/activity/SynchroExec.hpp"
+#include "src/kernel/activity/SynchroComm.hpp"
+#include "src/kernel/activity/SynchroSleep.hpp"
+#include "src/kernel/activity/SynchroIo.hpp"
+#include "src/kernel/activity/SynchroRaw.hpp"
 
 #if HAVE_MC
 #include "src/mc/mc_private.h"
 XBT_LOG_NEW_CATEGORY(simix, "All SIMIX categories");
 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_kernel, simix, "Logging specific to SIMIX (kernel)");
 
-smx_global_t simix_global = NULL;
-static xbt_heap_t simix_timers = NULL;
+std::unique_ptr<simgrid::simix::Global> simix_global;
+static xbt_heap_t simix_timers = nullptr;
 
 /** @brief Timer datatype */
 typedef struct s_smx_timer {
-  double date;
-  void(* func)(void*);
-  void* args;
+  double date = 0.0;
+  simgrid::xbt::Task<void()> callback;
+
+  s_smx_timer() {}
+  s_smx_timer(double date, simgrid::xbt::Task<void()> callback)
+    : date(date), callback(std::move(callback)) {}
 } s_smx_timer_t;
 
-void (*SMPI_switch_data_segment)(int) = NULL;
+void (*SMPI_switch_data_segment)(int) = nullptr;
 
 int _sg_do_verbose_exit = 1;
 static void inthandler(int ignored)
@@ -122,11 +132,11 @@ static void install_segvhandler(void)
   if (!(old_stack.ss_flags & SS_DISABLE)) {
     XBT_DEBUG("An alternate stack was already installed (sp=%p, size=%zd, flags=%x). Restore it.",
               old_stack.ss_sp, old_stack.ss_size, old_stack.ss_flags);
-    sigaltstack(&old_stack, NULL);
+    sigaltstack(&old_stack, nullptr);
   }
 
   struct sigaction action, old_action;
-  action.sa_sigaction = segvhandler;
+  action.sa_sigaction = &segvhandler;
   action.sa_flags = SA_ONSTACK | SA_RESETHAND | SA_SIGINFO;
   sigemptyset(&action.sa_mask);
 
@@ -138,7 +148,7 @@ static void install_segvhandler(void)
     XBT_DEBUG("A signal handler was already installed for SIGSEGV (%p). Restore it.",
              (old_action.sa_flags & SA_SIGINFO) ?
              (void*)old_action.sa_sigaction : (void*)old_action.sa_handler);
-    sigaction(SIGSEGV, &old_action, NULL);
+    sigaction(SIGSEGV, &old_action, nullptr);
   }
 }
 
@@ -152,7 +162,7 @@ double SIMIX_timer_next(void)
 
 static void kill_process(smx_process_t process)
 {
-  SIMIX_process_kill(process, NULL);
+  SIMIX_process_kill(process, nullptr);
 }
 
 static std::function<void()> maestro_code;
@@ -189,20 +199,17 @@ void SIMIX_global_init(int *argc, char **argv)
 #endif
 
   if (!simix_global) {
-    simix_global = xbt_new0(s_smx_global_t, 1);
+    simix_global = std::unique_ptr<simgrid::simix::Global>(new simgrid::simix::Global());
 
-    simgrid::simix::Process proc;
-    simix_global->process_to_run = xbt_dynar_new(sizeof(smx_process_t), NULL);
-    simix_global->process_that_ran = xbt_dynar_new(sizeof(smx_process_t), NULL);
+    simgrid::simix::ActorImpl proc;
+    simix_global->process_to_run = xbt_dynar_new(sizeof(smx_process_t), nullptr);
+    simix_global->process_that_ran = xbt_dynar_new(sizeof(smx_process_t), nullptr);
     simix_global->process_list = xbt_swag_new(xbt_swag_offset(proc, process_hookup));
     simix_global->process_to_destroy = xbt_swag_new(xbt_swag_offset(proc, destroy_hookup));
-
-    simix_global->maestro_process = NULL;
-    simix_global->registered_functions = xbt_dict_new_homogeneous(NULL);
-
-    simix_global->create_process_function = SIMIX_process_create;
-    simix_global->kill_process_function = kill_process;
-    simix_global->cleanup_process_function = SIMIX_process_cleanup;
+    simix_global->maestro_process = nullptr;
+    simix_global->create_process_function = &SIMIX_process_create;
+    simix_global->kill_process_function = &kill_process;
+    simix_global->cleanup_process_function = &SIMIX_process_cleanup;
     simix_global->mutex = xbt_os_mutex_init();
 
     surf_init(argc, argv);      /* Initialize SURF structures */
@@ -212,10 +219,6 @@ void SIMIX_global_init(int *argc, char **argv)
     // a context object with the current context mestro):
     simgrid::simix::create_maestro(maestro_code);
 
-    /* context exception handlers */
-    __xbt_running_ctx_fetch = SIMIX_process_get_running_context;
-    __xbt_ex_terminate = SIMIX_process_exception_terminate;
-
     /* Prepare to display some more info when dying on Ctrl-C pressing */
     signal(SIGINT, inthandler);
 
@@ -234,16 +237,17 @@ void SIMIX_global_init(int *argc, char **argv)
       const char* name = storage->getName();
       // TODO, create sg_storage_by_name
       sg_storage_t s = xbt_lib_get_elm_or_null(storage_lib, name);
-      xbt_assert(s != NULL, "Storage not found for name %s", name);
+      xbt_assert(s != nullptr, "Storage not found for name %s", name);
 
-      SIMIX_storage_create(name, s, NULL);
+      SIMIX_storage_create(name, s, nullptr);
     });
 
     SIMIX_STORAGE_LEVEL = xbt_lib_add_level(storage_lib, SIMIX_storage_destroy);
   }
-  if (!simix_timers) {
-    simix_timers = xbt_heap_new(8, &free);
-  }
+  if (!simix_timers)
+    simix_timers = xbt_heap_new(8, [](void* p) {
+      delete static_cast<smx_timer_t>(p);
+    });
 
   if (xbt_cfg_get_boolean("clean-atexit"))
     atexit(SIMIX_clean);
@@ -262,6 +266,18 @@ int smx_cleaned = 0;
 void SIMIX_clean(void)
 {
   if (smx_cleaned) return; // to avoid double cleaning by java and C
+
+#if HAVE_SMPI
+  if (SIMIX_process_count()>0){
+    if(smpi_process_initialized()){
+      xbt_die("Process exited without calling MPI_Finalize - Killing simulation");
+    }else{
+      XBT_WARN("Process called exit when leaving - Skipping cleanups");
+      return;
+    }
+  }
+#endif
+
   smx_cleaned = 1;
   XBT_DEBUG("SIMIX_clean called. Simulation's over.");
   if (!xbt_dynar_is_empty(simix_global->process_to_run) && SIMIX_get_clock() == 0.0) {
@@ -277,37 +293,30 @@ void SIMIX_clean(void)
   SIMIX_mailbox_exit();
 
   xbt_heap_free(simix_timers);
-  simix_timers = NULL;
+  simix_timers = nullptr;
   /* Free the remaining data structures */
   xbt_dynar_free(&simix_global->process_to_run);
   xbt_dynar_free(&simix_global->process_that_ran);
   xbt_swag_free(simix_global->process_to_destroy);
   xbt_swag_free(simix_global->process_list);
-  simix_global->process_list = NULL;
-  simix_global->process_to_destroy = NULL;
-  xbt_dict_free(&(simix_global->registered_functions));
+  simix_global->process_list = nullptr;
+  simix_global->process_to_destroy = nullptr;
 
   xbt_os_mutex_destroy(simix_global->mutex);
-  simix_global->mutex = NULL;
+  simix_global->mutex = nullptr;
 
   /* Let's free maestro now */
   delete simix_global->maestro_process->context;
-  xbt_free(simix_global->maestro_process->running_ctx);
-  xbt_free(simix_global->maestro_process);
-  simix_global->maestro_process = NULL;
-
-  /* Restore the default exception setup */
-  __xbt_running_ctx_fetch = &__xbt_ex_ctx_default;
-  __xbt_ex_terminate = &__xbt_ex_terminate_default;
+  simix_global->maestro_process->context = nullptr;
+  delete simix_global->maestro_process;
+  simix_global->maestro_process = nullptr;
 
   /* Finish context module and SURF */
   SIMIX_context_mod_exit();
 
   surf_exit();
 
-  xbt_free(simix_global);
-  simix_global = NULL;
-
+  simix_global = nullptr;
   return;
 }
 
@@ -340,6 +349,76 @@ static int process_syscall_color(void *p)
   }
 }
 
+/** Wake up all processes waiting for a Surf action to finish */
+static void SIMIX_wake_processes()
+{
+  unsigned int iter;
+  surf_model_t model;
+  surf_action_t action;
+
+  xbt_dynar_foreach(all_existing_models, iter, model) {
+    XBT_DEBUG("Handling the processes whose action failed (if any)");
+    while ((action = surf_model_extract_failed_action_set(model))) {
+      XBT_DEBUG("   Handling Action %p",action);
+      SIMIX_simcall_exit((smx_synchro_t) action->getData());
+    }
+    XBT_DEBUG("Handling the processes whose action terminated normally (if any)");
+    while ((action = surf_model_extract_done_action_set(model))) {
+      XBT_DEBUG("   Handling Action %p",action);
+      if (action->getData() == nullptr)
+        XBT_DEBUG("probably vcpu's action %p, skip", action);
+      else
+        SIMIX_simcall_exit((smx_synchro_t) action->getData());
+    }
+  }
+}
+
+/** Handle any pending timer */
+static bool SIMIX_execute_timers()
+{
+  bool result = false;
+  while (xbt_heap_size(simix_timers) > 0 && SIMIX_get_clock() >= SIMIX_timer_next()) {
+    result = true;
+     //FIXME: make the timers being real callbacks
+     // (i.e. provide dispatchers that read and expand the args)
+     smx_timer_t timer = (smx_timer_t) xbt_heap_pop(simix_timers);
+     try {
+       timer->callback();
+     }
+     catch(...) {
+       xbt_die("Exception throwed ouf of timer callback");
+     }
+     delete timer;
+  }
+  return result;
+}
+
+/** Execute all the tasks that are queued
+ *
+ *  e.g. `.then()` callbacks of futures.
+ **/
+static bool SIMIX_execute_tasks()
+{
+  xbt_assert(simix_global->tasksTemp.empty());
+
+  if (simix_global->tasks.empty())
+    return false;
+
+  using std::swap;
+  do {
+    // We don't want the callbacks to modify the vector we are iterating over:
+    swap(simix_global->tasks, simix_global->tasksTemp);
+
+    // Execute all the queued tasks:
+    for (auto& task : simix_global->tasksTemp)
+      task();
+
+    simix_global->tasksTemp.clear();
+  } while (!simix_global->tasks.empty());
+
+  return true;
+}
+
 /**
  * \ingroup SIMIX_API
  * \brief Run the main simulation loop.
@@ -353,14 +432,13 @@ void SIMIX_run(void)
 
   double time = 0;
   smx_process_t process;
-  surf_action_t action;
-  smx_timer_t timer;
-  surf_model_t model;
-  unsigned int iter;
 
   do {
     XBT_DEBUG("New Schedule Round; size(queue)=%lu",
         xbt_dynar_length(simix_global->process_to_run));
+
+    SIMIX_execute_tasks();
+
     while (!xbt_dynar_is_empty(simix_global->process_to_run)) {
       XBT_DEBUG("New Sub-Schedule Round; size(queue)=%lu",
               xbt_dynar_length(simix_global->process_to_run));
@@ -426,27 +504,18 @@ void SIMIX_run(void)
        *   That would thus be a pure waste of time.
        */
 
+      unsigned int iter;
       xbt_dynar_foreach(simix_global->process_that_ran, iter, process) {
         if (process->simcall.call != SIMCALL_NONE) {
           SIMIX_simcall_handle(&process->simcall, 0);
         }
       }
-      /* Wake up all processes waiting for a Surf action to finish */
-      xbt_dynar_foreach(all_existing_models, iter, model) {
-        XBT_DEBUG("Handling process whose action failed");
-        while ((action = surf_model_extract_failed_action_set(model))) {
-          XBT_DEBUG("   Handling Action %p",action);
-          SIMIX_simcall_exit((smx_synchro_t) action->getData());
-        }
-        XBT_DEBUG("Handling process whose action terminated normally");
-        while ((action = surf_model_extract_done_action_set(model))) {
-          XBT_DEBUG("   Handling Action %p",action);
-          if (action->getData() == NULL)
-            XBT_DEBUG("probably vcpu's action %p, skip", action);
-          else
-            SIMIX_simcall_exit((smx_synchro_t) action->getData());
-        }
-      }
+
+      SIMIX_execute_tasks();
+      do {
+        SIMIX_wake_processes();
+      } while (SIMIX_execute_tasks());
+
     }
 
     time = SIMIX_timer_next();
@@ -455,39 +524,23 @@ void SIMIX_run(void)
       time = surf_solve(time);
       XBT_DEBUG("Moving time ahead : %g", time);
     }
+
     /* Notify all the hosts that have failed */
     /* FIXME: iterate through the list of failed host and mark each of them */
     /* as failed. On each host, signal all the running processes with host_fail */
 
-    /* Handle any pending timer */
-    while (xbt_heap_size(simix_timers) > 0 && SIMIX_get_clock() >= SIMIX_timer_next()) {
-       //FIXME: make the timers being real callbacks
-       // (i.e. provide dispatchers that read and expand the args)
-       timer = (smx_timer_t) xbt_heap_pop(simix_timers);
-       if (timer->func)
-         timer->func(timer->args);
-       xbt_free(timer);
-    }
-
-    /* Wake up all processes waiting for a Surf action to finish */
-    xbt_dynar_foreach(all_existing_models, iter, model) {
-      XBT_DEBUG("Handling process whose action failed");
-      while ((action = surf_model_extract_failed_action_set(model))) {
-        XBT_DEBUG("   Handling Action %p",action);
-        SIMIX_simcall_exit((smx_synchro_t) action->getData());
-      }
-      XBT_DEBUG("Handling process whose action terminated normally");
-      while ((action = surf_model_extract_done_action_set(model))) {
-        XBT_DEBUG("   Handling Action %p",action);
-        if (action->getData() == NULL)
-          XBT_DEBUG("probably vcpu's action %p, skip", action);
-        else
-          SIMIX_simcall_exit((smx_synchro_t) action->getData());
-      }
-    }
+    // Execute timers and tasks until there isn't anything to be done:
+    bool again = false;
+    do {
+      again = SIMIX_execute_timers();
+      if (SIMIX_execute_tasks())
+        again = true;
+      SIMIX_wake_processes();
+    } while (again);
 
     /* Autorestart all process */
-    char *hostname = NULL;
+    char *hostname = nullptr;
+    unsigned int iter;
     xbt_dynar_foreach(host_that_restart,iter,hostname) {
       XBT_INFO("Restart processes on host: %s",hostname);
       SIMIX_host_autorestart(sg_host_by_name(hostname));
@@ -497,7 +550,6 @@ void SIMIX_run(void)
     /* Clean processes to destroy */
     SIMIX_process_empty_trash();
 
-
     XBT_DEBUG("### time %f, empty %d", time, xbt_dynar_is_empty(simix_global->process_to_run));
 
   } while (time != -1.0 || !xbt_dynar_is_empty(simix_global->process_to_run));
@@ -521,22 +573,26 @@ void SIMIX_run(void)
  *   \param arg Parameters of the function
  *
  */
-smx_timer_t SIMIX_timer_set(double date, void (*function)(void*), void *arg)
+smx_timer_t SIMIX_timer_set(double date, void (*callback)(void*), void *arg)
 {
-  smx_timer_t timer = xbt_new0(s_smx_timer_t, 1);
+  smx_timer_t timer = new s_smx_timer_t(date, [=](){ callback(arg); });
+  xbt_heap_push(simix_timers, timer, date);
+  return timer;
+}
 
-  timer->date = date;
-  timer->func = function;
-  timer->args = arg;
+smx_timer_t SIMIX_timer_set(double date, simgrid::xbt::Task<void()> callback)
+{
+  smx_timer_t timer = new s_smx_timer_t(date, std::move(callback));
   xbt_heap_push(simix_timers, timer, date);
   return timer;
 }
+
 /** @brief cancels a timer that was added earlier */
 void SIMIX_timer_remove(smx_timer_t timer) {
   xbt_heap_rm_elm(simix_timers, timer, timer->date);
 }
 
-/** @brief Returns the date at which the timer will trigger (or 0 if NULL timer) */
+/** @brief Returns the date at which the timer will trigger (or 0 if nullptr timer) */
 double SIMIX_timer_get_date(smx_timer_t timer) {
   return timer?timer->date:0;
 }
@@ -549,8 +605,7 @@ double SIMIX_timer_get_date(smx_timer_t timer) {
  * to call SIMIX_process_create().
  * \param function create process function
  */
-void SIMIX_function_register_process_create(smx_creation_func_t
-                                                       function)
+void SIMIX_function_register_process_create(smx_creation_func_t function)
 {
   simix_global->create_process_function = function;
 }
@@ -586,11 +641,11 @@ void SIMIX_function_register_process_cleanup(void_pfn_smxprocess_t
 
 void SIMIX_display_process_status(void)
 {
-  if (simix_global->process_list == NULL) {
+  if (simix_global->process_list == nullptr) {
     return;
   }
 
-  smx_process_t process = NULL;
+  smx_process_t process = nullptr;
   int nbprocess = xbt_swag_size(simix_global->process_list);
 
   XBT_INFO("%d processes are still running, waiting for something.", nbprocess);
@@ -603,19 +658,19 @@ void SIMIX_display_process_status(void)
 
       const char* synchro_description = "unknown";
 
-      if (dynamic_cast<simgrid::simix::Exec*>(process->waiting_synchro) != nullptr)
+      if (dynamic_cast<simgrid::kernel::activity::Exec*>(process->waiting_synchro) != nullptr)
         synchro_description = "execution";
 
-      if (dynamic_cast<simgrid::simix::Comm*>(process->waiting_synchro) != nullptr)
+      if (dynamic_cast<simgrid::kernel::activity::Comm*>(process->waiting_synchro) != nullptr)
         synchro_description = "communication";
 
-      if (dynamic_cast<simgrid::simix::Sleep*>(process->waiting_synchro) != nullptr)
+      if (dynamic_cast<simgrid::kernel::activity::Sleep*>(process->waiting_synchro) != nullptr)
         synchro_description = "sleeping";
 
-      if (dynamic_cast<simgrid::simix::Raw*>(process->waiting_synchro) != nullptr)
+      if (dynamic_cast<simgrid::kernel::activity::Raw*>(process->waiting_synchro) != nullptr)
         synchro_description = "synchronization";
 
-      if (dynamic_cast<simgrid::simix::Io*>(process->waiting_synchro) != nullptr)
+      if (dynamic_cast<simgrid::kernel::activity::Io*>(process->waiting_synchro) != nullptr)
         synchro_description = "I/O";
 
 
@@ -631,12 +686,12 @@ void SIMIX_display_process_status(void)
 */
 
       XBT_INFO("Process %lu (%s@%s): waiting for %s synchro %p (%s) in state %d to finish",
-          process->pid, process->name, sg_host_get_name(process->host),
+          process->pid, process->name.c_str(), sg_host_get_name(process->host),
           synchro_description, process->waiting_synchro,
-          process->waiting_synchro->name, (int)process->waiting_synchro->state);
+          process->waiting_synchro->name.c_str(), (int)process->waiting_synchro->state);
     }
     else {
-      XBT_INFO("Process %lu (%s@%s)", process->pid, process->name, sg_host_get_name(process->host));
+      XBT_INFO("Process %lu (%s@%s)", process->pid, process->name.c_str(), sg_host_get_name(process->host));
     }
   }
 }
@@ -651,5 +706,5 @@ xbt_dict_t SIMIX_asr_get_properties(const char *name)
 
 int SIMIX_is_maestro()
 {
-  return simix_global==NULL /*SimDag*/|| SIMIX_process_self() == simix_global->maestro_process;
+  return simix_global==nullptr /*SimDag*/|| SIMIX_process_self() == simix_global->maestro_process;
 }