src/kernel/actor/ActorImpl.cpp

   1 /* Copyright (c) 2007-2021. The SimGrid Team. All rights reserved.          */
   2
   3 /* This program is free software; you can redistribute it and/or modify it
   4  * under the terms of the license (GNU LGPL) which comes with this package. */
   5
   6 #include "mc/mc.h"
   7 #include "simgrid/Exception.hpp"
   8 #include "simgrid/s4u/Actor.hpp"
   9 #include "simgrid/s4u/Exec.hpp"
  10 #include "src/kernel/activity/CommImpl.hpp"
  11 #include "src/kernel/activity/ExecImpl.hpp"
  12 #include "src/kernel/activity/IoImpl.hpp"
  13 #include "src/kernel/activity/SleepImpl.hpp"
  14 #include "src/kernel/activity/SynchroRaw.hpp"
  15 #include "src/mc/mc_replay.hpp"
  16 #include "src/mc/remote/AppSide.hpp"
  17 #include "src/simix/smx_private.hpp"
  18 #include "src/surf/HostImpl.hpp"
  19 #include "src/surf/cpu_interface.hpp"
  20
  21 #include <boost/core/demangle.hpp>
  22 #include <boost/range/algorithm.hpp>
  23 #include <utility>
  24
  25 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_process, simix, "Logging specific to SIMIX (process)");
  26
  27 /**
  28  * @brief Returns the current agent.
  29  *
  30  * This functions returns the currently running SIMIX process.
  31  *
  32  * @return The SIMIX process
  33  */
  34 smx_actor_t SIMIX_process_self()
  35 {
  36   return simgrid::kernel::actor::ActorImpl::self();
  37 }
  38
  39 namespace simgrid {
  40 namespace kernel {
  41 namespace actor {
  42
  43 static unsigned long maxpid = 0;
  44 unsigned long get_maxpid()
  45 {
  46   return maxpid;
  47 }
  48 void* get_maxpid_addr()
  49 {
  50   return &maxpid;
  51 }
  52 ActorImpl* ActorImpl::by_pid(aid_t pid)
  53 {
  54   auto item = simix_global->process_list.find(pid);
  55   if (item != simix_global->process_list.end())
  56     return item->second;
  57
  58   // Search the trash
  59   for (auto& a : simix_global->actors_to_destroy)
  60     if (a.get_pid() == pid)
  61       return &a;
  62   return nullptr; // Not found, even in the trash
  63 }
  64
  65 ActorImpl* ActorImpl::self()
  66 {
  67   const context::Context* self_context = context::Context::self();
  68
  69   return (self_context != nullptr) ? self_context->get_actor() : nullptr;
  70 }
  71
  72 ActorImpl::ActorImpl(xbt::string name, s4u::Host* host) : host_(host), name_(std::move(name)), piface_(this)
  73 {
  74   pid_            = maxpid++;
  75   simcall_.issuer_ = this;
  76   stacksize_      = smx_context_stack_size;
  77 }
  78
  79 ActorImpl::~ActorImpl()
  80 {
  81   if (simix_global != nullptr && this != simix_global->maestro_)
  82     s4u::Actor::on_destruction(*get_ciface());
  83 }
  84
  85 /* Become an actor in the simulation
  86  *
  87  * Currently this can only be called by the main thread (once) and only work with some thread factories
  88  * (currently ThreadContextFactory).
  89  *
  90  * In the future, it might be extended in order to attach other threads created by a third party library.
  91  */
  92
  93 ActorImplPtr ActorImpl::attach(const std::string& name, void* data, s4u::Host* host,
  94                                const std::unordered_map<std::string, std::string>* properties)
  95 {
  96   // This is mostly a copy/paste from create(), it'd be nice to share some code between those two functions.
  97
  98   XBT_DEBUG("Attach actor %s on host '%s'", name.c_str(), host->get_cname());
  99
 100   if (not host->is_on()) {
 101     XBT_WARN("Cannot attach actor '%s' on failed host '%s'", name.c_str(), host->get_cname());
 102     throw HostFailureException(XBT_THROW_POINT, "Cannot attach actor on failed host.");
 103   }
 104
 105   auto* actor = new ActorImpl(xbt::string(name), host);
 106   /* Actor data */
 107   actor->set_user_data(data);
 108   actor->code_ = nullptr;
 109
 110   XBT_VERB("Create context %s", actor->get_cname());
 111   xbt_assert(simix_global != nullptr, "simix is not initialized, please call MSG_init first");
 112   actor->context_.reset(simix_global->context_factory->attach(actor));
 113
 114   /* Add properties */
 115   if (properties != nullptr)
 116     actor->set_properties(*properties);
 117
 118   /* Add the actor to it's host actor list */
 119   host->pimpl_->add_actor(actor);
 120
 121   /* Now insert it in the global actor list and in the actors to run list */
 122   simix_global->process_list[actor->get_pid()] = actor;
 123   XBT_DEBUG("Inserting [%p] %s(%s) in the to_run list", actor, actor->get_cname(), host->get_cname());
 124   simix_global->actors_to_run.push_back(actor);
 125   intrusive_ptr_add_ref(actor);
 126
 127   auto* context = dynamic_cast<context::AttachContext*>(actor->context_.get());
 128   xbt_assert(nullptr != context, "Not a suitable context");
 129   context->attach_start();
 130
 131   /* The on_creation() signal must be delayed until there, where the pid and everything is set */
 132   s4u::Actor::on_creation(*actor->get_ciface());
 133
 134   return ActorImplPtr(actor);
 135 }
 136 /** @brief Detach an actor attached with `attach()`
 137  *
 138  *  This is called when the current actor has finished its job.
 139  *  Used in the main thread, it waits for the simulation to finish before returning. When it returns, the other
 140  *  simulated actors and the maestro are destroyed.
 141  */
 142 void ActorImpl::detach()
 143 {
 144   auto* context = dynamic_cast<context::AttachContext*>(context::Context::self());
 145   xbt_assert(context != nullptr, "Not a suitable context");
 146
 147   context->get_actor()->cleanup();
 148   context->attach_stop();
 149 }
 150
 151 void ActorImpl::cleanup_from_simix()
 152 {
 153   const std::lock_guard<std::mutex> lock(simix_global->mutex);
 154   simix_global->process_list.erase(pid_);
 155   if (host_ && host_actor_list_hook.is_linked())
 156     host_->pimpl_->remove_actor(this);
 157   if (not smx_destroy_list_hook.is_linked()) {
 158 #if SIMGRID_HAVE_MC
 159     xbt_dynar_push_as(simix_global->dead_actors_vector, ActorImpl*, this);
 160 #endif
 161     simix_global->actors_to_destroy.push_back(*this);
 162   }
 163 }
 164
 165 void ActorImpl::cleanup()
 166 {
 167   finished_ = true;
 168
 169   if (has_to_auto_restart() && not get_host()->is_on()) {
 170     XBT_DEBUG("Insert host %s to watched_hosts because it's off and %s needs to restart", get_host()->get_cname(),
 171               get_cname());
 172     watched_hosts().insert(get_host()->get_name());
 173   }
 174
 175   if (on_exit) {
 176     // Execute the termination callbacks
 177     bool failed = context_->wannadie();
 178     for (auto exit_fun = on_exit->crbegin(); exit_fun != on_exit->crend(); ++exit_fun)
 179       (*exit_fun)(failed);
 180     on_exit.reset();
 181   }
 182   undaemonize();
 183
 184   /* cancel non-blocking activities */
 185   for (auto activity : activities_)
 186     activity->cancel();
 187   activities_.clear();
 188
 189   XBT_DEBUG("%s@%s(%ld) should not run anymore", get_cname(), get_host()->get_cname(), get_pid());
 190
 191   if (this == simix_global->maestro_) /* Do not cleanup maestro */
 192     return;
 193
 194   XBT_DEBUG("Cleanup actor %s (%p), waiting synchro %p", get_cname(), this, waiting_synchro_.get());
 195
 196   /* Unregister associated timers if any */
 197   if (kill_timer_ != nullptr) {
 198     kill_timer_->remove();
 199     kill_timer_ = nullptr;
 200   }
 201   if (simcall_.timeout_cb_) {
 202     simcall_.timeout_cb_->remove();
 203     simcall_.timeout_cb_ = nullptr;
 204   }
 205
 206   cleanup_from_simix();
 207
 208   context_->set_wannadie(false); // don't let the simcall's yield() do a Context::stop(), to avoid infinite loops
 209   actor::simcall([this] { s4u::Actor::on_termination(*get_ciface()); });
 210   context_->set_wannadie();
 211 }
 212
 213 void ActorImpl::exit()
 214 {
 215   context_->set_wannadie();
 216   suspended_          = false;
 217   exception_          = nullptr;
 218
 219   /* destroy the blocking synchro if any */
 220   if (waiting_synchro_ != nullptr) {
 221     waiting_synchro_->cancel();
 222     waiting_synchro_->state_ = activity::State::FAILED;
 223
 224     activity::ExecImplPtr exec = boost::dynamic_pointer_cast<activity::ExecImpl>(waiting_synchro_);
 225     activity::CommImplPtr comm = boost::dynamic_pointer_cast<activity::CommImpl>(waiting_synchro_);
 226
 227     if (exec != nullptr) {
 228       exec->clean_action();
 229     } else if (comm != nullptr) {
 230       // Remove first occurrence of &actor->simcall:
 231       auto i = boost::range::find(waiting_synchro_->simcalls_, &simcall_);
 232       if (i != waiting_synchro_->simcalls_.end())
 233         waiting_synchro_->simcalls_.remove(&simcall_);
 234     } else {
 235       activity::ActivityImplPtr(waiting_synchro_)->finish();
 236     }
 237
 238     activities_.remove(waiting_synchro_);
 239     waiting_synchro_ = nullptr;
 240   }
 241
 242   // Forcefully kill the actor if its host is turned off. Not a HostFailureException because you should not survive that
 243   this->throw_exception(std::make_exception_ptr(ForcefulKillException(host_->is_on() ? "exited" : "host failed")));
 244 }
 245
 246 void ActorImpl::kill(ActorImpl* actor) const
 247 {
 248   xbt_assert(actor != simix_global->maestro_, "Killing maestro is a rather bad idea");
 249   if (actor->finished_) {
 250     XBT_DEBUG("Ignoring request to kill actor %s@%s that is already dead", actor->get_cname(),
 251               actor->host_->get_cname());
 252     return;
 253   }
 254
 255   XBT_DEBUG("Actor '%s'@%s is killing actor '%s'@%s", get_cname(), host_ ? host_->get_cname() : "", actor->get_cname(),
 256             actor->host_ ? actor->host_->get_cname() : "");
 257
 258   actor->exit();
 259
 260   if (actor == this) {
 261     XBT_DEBUG("Go on, this is a suicide,");
 262   } else if (std::find(begin(simix_global->actors_to_run), end(simix_global->actors_to_run), actor) !=
 263              end(simix_global->actors_to_run)) {
 264     XBT_DEBUG("Actor %s is already in the to_run list", actor->get_cname());
 265   } else {
 266     XBT_DEBUG("Inserting %s in the to_run list", actor->get_cname());
 267     simix_global->actors_to_run.push_back(actor);
 268   }
 269 }
 270
 271 void ActorImpl::kill_all() const
 272 {
 273   for (auto const& kv : simix_global->process_list)
 274     if (kv.second != this)
 275       this->kill(kv.second);
 276 }
 277
 278 void ActorImpl::set_kill_time(double kill_time)
 279 {
 280   if (kill_time <= SIMIX_get_clock())
 281     return;
 282   XBT_DEBUG("Set kill time %f for actor %s@%s", kill_time, get_cname(), host_->get_cname());
 283   kill_timer_ = simix::Timer::set(kill_time, [this] {
 284     this->exit();
 285     kill_timer_ = nullptr;
 286   });
 287 }
 288
 289 double ActorImpl::get_kill_time() const
 290 {
 291   return kill_timer_ ? kill_timer_->get_date() : 0;
 292 }
 293
 294 void ActorImpl::yield()
 295 {
 296   XBT_DEBUG("Yield actor '%s'", get_cname());
 297
 298   /* Go into sleep and return control to maestro */
 299   context_->suspend();
 300
 301   /* Ok, maestro returned control to us */
 302   XBT_DEBUG("Control returned to me: '%s'", get_cname());
 303
 304   if (context_->wannadie()) {
 305     XBT_DEBUG("Actor %s@%s is dead", get_cname(), host_->get_cname());
 306     context_->stop();
 307     THROW_IMPOSSIBLE;
 308   }
 309
 310   if (suspended_) {
 311     XBT_DEBUG("Hey! I'm suspended.");
 312
 313     xbt_assert(exception_ == nullptr, "Gasp! This exception may be lost by subsequent calls.");
 314
 315     if (waiting_synchro_ != nullptr) // Not sure of when this will happen. Maybe when suspending early in the SR when a
 316       waiting_synchro_->suspend();   // waiting_synchro was terminated
 317
 318     yield(); // Yield back to maestro without proceeding with my execution. I'll get rescheduled by resume()
 319   }
 320
 321   if (exception_ != nullptr) {
 322     XBT_DEBUG("Wait, maestro left me an exception");
 323     std::exception_ptr exception = std::move(exception_);
 324     exception_                   = nullptr;
 325     try {
 326       std::rethrow_exception(std::move(exception));
 327     } catch (const simgrid::Exception& e) {
 328       e.rethrow_nested(XBT_THROW_POINT, boost::core::demangle(typeid(e).name()) + " raised in kernel mode.");
 329     }
 330   }
 331
 332   if (SMPI_switch_data_segment && not finished_) {
 333     SMPI_switch_data_segment(get_iface());
 334   }
 335 }
 336
 337 /** This actor will be terminated automatically when the last non-daemon actor finishes */
 338 void ActorImpl::daemonize()
 339 {
 340   if (not daemon_) {
 341     daemon_ = true;
 342     simix_global->daemons.push_back(this);
 343   }
 344 }
 345
 346 void ActorImpl::undaemonize()
 347 {
 348   if (daemon_) {
 349     auto& vect = simix_global->daemons;
 350     auto it    = std::find(vect.begin(), vect.end(), this);
 351     xbt_assert(it != vect.end(), "The dying daemon is not a daemon after all. Please report that bug.");
 352     /* Don't move the whole content since we don't really care about the order */
 353
 354     std::swap(*it, vect.back());
 355     vect.pop_back();
 356     daemon_ = false;
 357   }
 358 }
 359
 360 s4u::Actor* ActorImpl::restart()
 361 {
 362   xbt_assert(this != simix_global->maestro_, "Restarting maestro is not supported");
 363
 364   XBT_DEBUG("Restarting actor %s on %s", get_cname(), host_->get_cname());
 365
 366   // retrieve the arguments of the old actor
 367   ProcessArg arg = ProcessArg(host_, this);
 368
 369   // kill the old actor
 370   context::Context::self()->get_actor()->kill(this);
 371
 372   // start the new actor
 373   ActorImplPtr actor = ActorImpl::create(arg.name, arg.code, arg.data, arg.host, arg.properties.get(), nullptr);
 374   *actor->on_exit = std::move(*arg.on_exit);
 375   actor->set_kill_time(arg.kill_time);
 376   actor->set_auto_restart(arg.auto_restart);
 377
 378   return actor->get_ciface();
 379 }
 380
 381 void ActorImpl::suspend()
 382 {
 383   if (suspended_) {
 384     XBT_DEBUG("Actor '%s' is already suspended", get_cname());
 385     return;
 386   }
 387
 388   suspended_ = true;
 389
 390   /* If the suspended actor is waiting on a sync, suspend its synchronization.
 391    * Otherwise, it will suspend itself when scheduled, ie, very soon. */
 392   if (waiting_synchro_ != nullptr)
 393     waiting_synchro_->suspend();
 394 }
 395
 396 void ActorImpl::resume()
 397 {
 398   XBT_IN("actor = %p", this);
 399
 400   if (context_->wannadie()) {
 401     XBT_VERB("Ignoring request to suspend an actor that is currently dying.");
 402     return;
 403   }
 404
 405   if (not suspended_)
 406     return;
 407   suspended_ = false;
 408
 409   /* resume the activity that was blocking the resumed actor. */
 410   if (waiting_synchro_)
 411     waiting_synchro_->resume();
 412   else // Reschedule the actor if it was forcefully unscheduled in yield()
 413     simix_global->actors_to_run.push_back(this);
 414
 415   XBT_OUT();
 416 }
 417
 418 activity::ActivityImplPtr ActorImpl::join(const ActorImpl* actor, double timeout)
 419 {
 420   activity::ActivityImplPtr sleep = this->sleep(timeout);
 421   actor->on_exit->emplace_back([sleep](bool) {
 422     if (sleep->surf_action_)
 423       sleep->surf_action_->finish(resource::Action::State::FINISHED);
 424   });
 425   return sleep;
 426 }
 427
 428 activity::ActivityImplPtr ActorImpl::sleep(double duration)
 429 {
 430   if (not host_->is_on())
 431     throw_exception(std::make_exception_ptr(HostFailureException(
 432         XBT_THROW_POINT, std::string("Host ") + host_->get_cname() + " failed, you cannot sleep there.")));
 433
 434   auto sleep = new activity::SleepImpl();
 435   sleep->set_name("sleep").set_host(host_).set_duration(duration).start();
 436   return activity::SleepImplPtr(sleep);
 437 }
 438
 439 void ActorImpl::throw_exception(std::exception_ptr e)
 440 {
 441   exception_ = e;
 442
 443   if (suspended_)
 444     resume();
 445
 446   /* cancel the blocking synchro if any */
 447   if (waiting_synchro_) {
 448     waiting_synchro_->cancel();
 449     activities_.remove(waiting_synchro_);
 450     waiting_synchro_ = nullptr;
 451   }
 452 }
 453
 454 void ActorImpl::simcall_answer()
 455 {
 456   if (this != simix_global->maestro_) {
 457     XBT_DEBUG("Answer simcall %s (%d) issued by %s (%p)", SIMIX_simcall_name(simcall_.call_), (int)simcall_.call_,
 458               get_cname(), this);
 459     xbt_assert(simcall_.call_ != simix::Simcall::NONE);
 460     simcall_.call_ = simix::Simcall::NONE;
 461     xbt_assert(not XBT_LOG_ISENABLED(simix_process, xbt_log_priority_debug) ||
 462                    std::find(begin(simix_global->actors_to_run), end(simix_global->actors_to_run), this) ==
 463                        end(simix_global->actors_to_run),
 464                "Actor %p should not exist in actors_to_run!", this);
 465     simix_global->actors_to_run.push_back(this);
 466   }
 467 }
 468
 469 void ActorImpl::set_host(s4u::Host* dest)
 470 {
 471   host_->pimpl_->remove_actor(this);
 472   host_ = dest;
 473   dest->pimpl_->add_actor(this);
 474 }
 475
 476 ActorImplPtr ActorImpl::init(const std::string& name, s4u::Host* host) const
 477 {
 478   auto* actor = new ActorImpl(xbt::string(name), host);
 479   actor->set_ppid(this->pid_);
 480
 481   intrusive_ptr_add_ref(actor);
 482   /* The on_creation() signal must be delayed until there, where the pid and everything is set */
 483   s4u::Actor::on_creation(*actor->get_ciface());
 484
 485   return ActorImplPtr(actor);
 486 }
 487
 488 ActorImpl* ActorImpl::start(const ActorCode& code)
 489 {
 490   xbt_assert(code && host_ != nullptr, "Invalid parameters");
 491
 492   if (not host_->is_on()) {
 493     XBT_WARN("Cannot launch actor '%s' on failed host '%s'", name_.c_str(), host_->get_cname());
 494     intrusive_ptr_release(this);
 495     throw HostFailureException(XBT_THROW_POINT, "Cannot start actor on failed host.");
 496   }
 497
 498   this->code_ = code;
 499   XBT_VERB("Create context %s", get_cname());
 500   context_.reset(simix_global->context_factory->create_context(ActorCode(code), this));
 501
 502   XBT_DEBUG("Start context '%s'", get_cname());
 503
 504   /* Add the actor to its host's actor list */
 505   host_->pimpl_->add_actor(this);
 506   simix_global->process_list[pid_] = this;
 507
 508   /* Now insert it in the global actor list and in the actor to run list */
 509   XBT_DEBUG("Inserting [%p] %s(%s) in the to_run list", this, get_cname(), host_->get_cname());
 510   simix_global->actors_to_run.push_back(this);
 511
 512   return this;
 513 }
 514
 515 ActorImplPtr ActorImpl::create(const std::string& name, const ActorCode& code, void* data, s4u::Host* host,
 516                                const std::unordered_map<std::string, std::string>* properties,
 517                                const ActorImpl* parent_actor)
 518 {
 519   XBT_DEBUG("Start actor %s@'%s'", name.c_str(), host->get_cname());
 520
 521   ActorImplPtr actor;
 522   if (parent_actor != nullptr)
 523     actor = parent_actor->init(xbt::string(name), host);
 524   else
 525     actor = self()->init(xbt::string(name), host);
 526
 527   /* actor data */
 528   actor->set_user_data(data);
 529
 530   /* Add properties */
 531   if (properties != nullptr)
 532     actor->set_properties(*properties);
 533
 534   actor->start(code);
 535
 536   return actor;
 537 }
 538
 539 void create_maestro(const std::function<void()>& code)
 540 {
 541   /* Create maestro actor and initialize it */
 542   auto* maestro = new ActorImpl(xbt::string(""), /*host*/ nullptr);
 543
 544   if (not code) {
 545     maestro->context_.reset(simix_global->context_factory->create_context(ActorCode(), maestro));
 546   } else {
 547     maestro->context_.reset(simix_global->context_factory->create_maestro(ActorCode(code), maestro));
 548   }
 549
 550   maestro->simcall_.issuer_     = maestro;
 551   simix_global->maestro_        = maestro;
 552 }
 553
 554 } // namespace actor
 555 } // namespace kernel
 556 } // namespace simgrid
 557
 558 int SIMIX_process_count() // XBT_ATTRIB_DEPRECATED_v329
 559 {
 560   return simix_global->process_list.size();
 561 }
 562
 563 void* SIMIX_process_self_get_data() // XBT_ATTRIB_DEPRECATED_v329
 564 {
 565   smx_actor_t self = simgrid::kernel::actor::ActorImpl::self();
 566
 567   if (self == nullptr) {
 568     return nullptr;
 569   }
 570   return self->get_user_data();
 571 }
 572
 573 void SIMIX_process_self_set_data(void* data) // XBT_ATTRIB_DEPRECATED_v329
 574 {
 575   simgrid::kernel::actor::ActorImpl::self()->set_user_data(data);
 576 }
 577
 578 /* needs to be public and without simcall because it is called
 579    by exceptions and logging events */
 580 const char* SIMIX_process_self_get_name()
 581 {
 582   return SIMIX_is_maestro() ? "maestro" : simgrid::kernel::actor::ActorImpl::self()->get_cname();
 583 }
 584
 585 /** @brief Returns the process from PID. */
 586 smx_actor_t SIMIX_process_from_PID(aid_t pid)
 587 {
 588   return simgrid::kernel::actor::ActorImpl::by_pid(pid);
 589 }
 590
 591 void SIMIX_process_on_exit(smx_actor_t actor,
 592                            const std::function<void(bool /*failed*/)>& fun) // XBT_ATTRIB_DEPRECATED_v329
 593 {
 594   xbt_assert(actor, "current process not found: are you in maestro context ?");
 595   actor->on_exit->emplace_back(fun);
 596 }
 597
 598 void simcall_process_set_data(smx_actor_t process, void* data) // XBT_ATTRIB_DEPRECATED_v329
 599 {
 600   simgrid::kernel::actor::simcall([process, data] { process->set_user_data(data); });
 601 }