allow one actor to kill another actor by PID

[simgrid.git] / src / s4u / s4u_actor.cpp

diff --git a/src/s4u/s4u_actor.cpp b/src/s4u/s4u_actor.cpp
index a437de0..39c8ab6 100644 (file)
--- a/src/s4u/s4u_actor.cpp
+++ b/src/s4u/s4u_actor.cpp
@@ -21,6 +21,9 @@ XBT_LOG_NEW_DEFAULT_CATEGORY(s4u_actor, "S4U actors");
 namespace simgrid {
 namespace s4u {
 
+simgrid::xbt::signal<void(simgrid::s4u::ActorPtr)> s4u::Actor::onCreation;
+simgrid::xbt::signal<void(simgrid::s4u::ActorPtr)> s4u::Actor::onDestruction;
+
 // ***** Actor creation *****
 ActorPtr Actor::self()
 {
@@ -74,9 +77,28 @@ void Actor::onExit(int_f_pvoid_pvoid_t fun, void* data)
   simcall_process_on_exit(pimpl_, fun, data);
 }
 
+/** @brief Moves the actor to another host
+ *
+ * If the actor is currently blocked on an execution activity, the activity is also
+ * migrated to the new host. If it's blocked on another kind of activity, an error is
+ * raised as the mandated code is not written yet. Please report that bug if you need it.
+ *
+ * Asynchronous activities started by the actor are not migrated automatically, so you have
+ * to take care of this yourself (only you knows which ones should be migrated).
+ */
 void Actor::migrate(Host* new_host)
 {
-  simgrid::simix::kernelImmediate([this, new_host]() { pimpl_->new_host = new_host; });
+  simgrid::simix::kernelImmediate([this, new_host]() {
+    if (pimpl_->waiting_synchro != nullptr) {
+      // The actor is blocked on an activity. If it's an exec, migrate it too.
+      // FIXME: implement the migration of other kind of activities
+      simgrid::kernel::activity::ExecImplPtr exec =
+          boost::dynamic_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_->waiting_synchro);
+      xbt_assert(exec.get() != nullptr, "We can only migrate blocked actors when they are blocked on executions.");
+      exec->migrate(new_host);
+    }
+    SIMIX_process_change_host(this->pimpl_, new_host);
+  });
 }
 
 s4u::Host* Actor::getHost()
@@ -89,6 +111,11 @@ void Actor::daemonize()
   simgrid::simix::kernelImmediate([this]() { pimpl_->daemonize(); });
 }
 
+bool Actor::isDaemon()
+{
+  return this->pimpl_->isDaemon();
+}
+
 const simgrid::xbt::string& Actor::getName() const
 {
   return this->pimpl_->getName();
@@ -111,12 +138,17 @@ aid_t Actor::getPpid()
 
 void Actor::suspend()
 {
+  if (TRACE_actor_is_enabled())
+    simgrid::instr::Container::byName(instr_pid(this))->getState("MSG_PROCESS_STATE")->pushEvent("suspend");
+
   simcall_process_suspend(pimpl_);
 }
 
 void Actor::resume()
 {
   simgrid::simix::kernelImmediate([this] { pimpl_->resume(); });
+  if (TRACE_actor_is_enabled())
+    simgrid::instr::Container::byName(instr_pid(this))->getState("MSG_PROCESS_STATE")->popEvent();
 }
 
 int Actor::isSuspended()
@@ -136,26 +168,28 @@ double Actor::getKillTime()
 
 void Actor::kill(aid_t pid)
 {
+  smx_actor_t killer  = SIMIX_process_self();
   smx_actor_t process = SIMIX_process_from_PID(pid);
   if(process != nullptr) {
-    simgrid::simix::kernelImmediate([process] { SIMIX_process_kill(process, process); });
+    simgrid::simix::kernelImmediate([killer, process] { SIMIX_process_kill(process, killer); });
   } else {
     std::ostringstream oss;
-    oss << "kill: ("<< pid <<") - No such process" << std::endl;
+    oss << "kill: (" << pid << ") - No such actor" << std::endl;
     throw std::runtime_error(oss.str());
   }
 }
 
-smx_actor_t Actor::getImpl() {
-  return pimpl_;
-}
-
 void Actor::kill() {
   smx_actor_t process = SIMIX_process_self();
   simgrid::simix::kernelImmediate(
       [this, process] { SIMIX_process_kill(pimpl_, (pimpl_ == simix_global->maestro_process) ? pimpl_ : process); });
 }
 
+smx_actor_t Actor::getImpl()
+{
+  return pimpl_;
+}
+
 // ***** Static functions *****
 
 ActorPtr Actor::byPid(aid_t pid)
@@ -198,22 +232,6 @@ Actor* Actor::restart()
   return simgrid::simix::kernelImmediate([this]() { return pimpl_->restart(); });
 }
 
-ExecPtr Actor::exec_init(double flops_amount)
-{
-  ExecPtr res        = ExecPtr(new Exec());
-  res->host_         = this->getHost();
-  res->flops_amount_ = flops_amount;
-  res->setRemains(flops_amount);
-  return res;
-}
-
-ExecPtr Actor::exec_async(double flops)
-{
-  ExecPtr res = exec_init(flops);
-  res->start();
-  return res;
-}
-
 // ***** this_actor *****
 
 namespace this_actor {
@@ -276,6 +294,22 @@ void parallel_execute(int host_nb, sg_host_t* host_list, double* flops_amount, d
   simcall_execution_wait(s);
 }
 
+ExecPtr exec_init(double flops_amount)
+{
+  ExecPtr res        = ExecPtr(new Exec());
+  res->host_         = getHost();
+  res->flops_amount_ = flops_amount;
+  res->setRemains(flops_amount);
+  return res;
+}
+
+ExecPtr exec_async(double flops)
+{
+  ExecPtr res = exec_init(flops);
+  res->start();
+  return res;
+}
+
 void* recv(MailboxPtr chan) // deprecated
 {
   return chan->get();
@@ -333,6 +367,10 @@ Host* getHost()
 
 void suspend()
 {
+  if (TRACE_actor_is_enabled())
+    instr::Container::byName(getName() + "-" + std::to_string(getPid()))
+        ->getState("MSG_PROCESS_STATE")
+        ->pushEvent("suspend");
   simcall_process_suspend(SIMIX_process_self());
 }
 
@@ -340,6 +378,9 @@ void resume()
 {
   smx_actor_t process = SIMIX_process_self();
   simgrid::simix::kernelImmediate([process] { process->resume(); });
+
+  if (TRACE_actor_is_enabled())
+    instr::Container::byName(getName() + "-" + std::to_string(getPid()))->getState("MSG_PROCESS_STATE")->popEvent();
 }
 
 bool isSuspended()
@@ -359,10 +400,13 @@ void onExit(int_f_pvoid_pvoid_t fun, void* data)
   simcall_process_on_exit(SIMIX_process_self(), fun, data);
 }
 
+/** @brief Moves the current actor to another host
+ *
+ * @see simgrid::s4u::Actor::migrate() for more information
+ */
 void migrate(Host* new_host)
 {
-  smx_actor_t process = SIMIX_process_self();
-  simgrid::simix::kernelImmediate([process, new_host] { process->new_host = new_host; });
+  SIMIX_process_self()->iface()->migrate(new_host);
 }
 }
 }