Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
21398a2f2c65904f4c2cc9dfb4d806f3d1fa5c48
[simgrid.git] / src / s4u / s4u_Actor.cpp
1 /* Copyright (c) 2006-2019. 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 "simgrid/Exception.hpp"
7 #include "simgrid/actor.h"
8 #include "simgrid/s4u/Actor.hpp"
9 #include "simgrid/s4u/Exec.hpp"
10 #include "simgrid/s4u/Host.hpp"
11 #include "simgrid/s4u/VirtualMachine.hpp"
12 #include "src/kernel/activity/ExecImpl.hpp"
13 #include "src/simix/smx_host_private.hpp"
14 #include "src/simix/smx_private.hpp"
15 #include "src/surf/HostImpl.hpp"
16
17 #include <algorithm>
18 #include <sstream>
19
20 XBT_LOG_NEW_DEFAULT_CATEGORY(s4u_actor, "S4U actors");
21
22 namespace simgrid {
23 namespace s4u {
24
25 simgrid::xbt::signal<void(simgrid::s4u::ActorPtr)> s4u::Actor::on_creation;
26 simgrid::xbt::signal<void(simgrid::s4u::ActorPtr)> s4u::Actor::on_suspend;
27 simgrid::xbt::signal<void(simgrid::s4u::ActorPtr)> s4u::Actor::on_resume;
28 simgrid::xbt::signal<void(simgrid::s4u::ActorPtr)> s4u::Actor::on_sleep;
29 simgrid::xbt::signal<void(simgrid::s4u::ActorPtr)> s4u::Actor::on_wake_up;
30 simgrid::xbt::signal<void(simgrid::s4u::ActorPtr)> s4u::Actor::on_migration_start;
31 simgrid::xbt::signal<void(simgrid::s4u::ActorPtr)> s4u::Actor::on_migration_end;
32 simgrid::xbt::signal<void(simgrid::s4u::ActorPtr)> s4u::Actor::on_destruction;
33
34 // ***** Actor creation *****
35 ActorPtr Actor::self()
36 {
37   smx_context_t self_context = simgrid::kernel::context::Context::self();
38   if (self_context == nullptr)
39     return simgrid::s4u::ActorPtr();
40
41   return self_context->get_actor()->iface();
42 }
43
44 ActorPtr Actor::create(std::string name, s4u::Host* host, std::function<void()> code)
45 {
46   simgrid::kernel::actor::ActorImpl* actor =
47       simcall_process_create(std::move(name), std::move(code), nullptr, host, nullptr);
48   return actor->iface();
49 }
50
51 ActorPtr Actor::create(std::string name, s4u::Host* host, const std::string& function, std::vector<std::string> args)
52 {
53   simgrid::simix::ActorCodeFactory& factory = SIMIX_get_actor_code_factory(function);
54   return create(std::move(name), host, factory(std::move(args)));
55 }
56
57 void intrusive_ptr_add_ref(Actor* actor)
58 {
59   intrusive_ptr_add_ref(actor->pimpl_);
60 }
61 void intrusive_ptr_release(Actor* actor)
62 {
63   intrusive_ptr_release(actor->pimpl_);
64 }
65
66 // ***** Actor methods *****
67
68 void Actor::join()
69 {
70   simcall_process_join(this->pimpl_, -1);
71 }
72
73 void Actor::join(double timeout)
74 {
75   simcall_process_join(this->pimpl_, timeout);
76 }
77
78 void Actor::set_auto_restart(bool autorestart)
79 {
80   simgrid::simix::simcall([this, autorestart]() {
81     xbt_assert(autorestart && not pimpl_->has_to_auto_restart()); // FIXME: handle all cases
82     pimpl_->set_auto_restart(autorestart);
83
84     simgrid::kernel::actor::ProcessArg* arg = new simgrid::kernel::actor::ProcessArg(pimpl_->get_host(), pimpl_);
85     XBT_DEBUG("Adding Process %s to the actors_at_boot_ list of Host %s", arg->name.c_str(), arg->host->get_cname());
86     pimpl_->get_host()->pimpl_->actors_at_boot_.emplace_back(arg);
87   });
88 }
89
90 void Actor::on_exit(int_f_pvoid_pvoid_t fun,
91                     void* data) /* deprecated: cleanup SIMIX_process_on_exit: change prototype of second parameter and
92                                    remove the last one */
93 {
94   simgrid::simix::simcall([this, fun, data] { SIMIX_process_on_exit(pimpl_, fun, data); });
95 }
96
97 void Actor::on_exit(std::function<void(bool /*failed*/)> const fun)
98 {
99   simgrid::simix::simcall(
100       [this, fun] { SIMIX_process_on_exit(pimpl_, [fun](int a, void* /*data*/) { fun(a != 0); }, nullptr); });
101 }
102
103 void Actor::migrate(Host* new_host)
104 {
105   s4u::Actor::on_migration_start(this);
106
107   simgrid::simix::simcall([this, new_host]() {
108     if (pimpl_->waiting_synchro != nullptr) {
109       // The actor is blocked on an activity. If it's an exec, migrate it too.
110       // FIXME: implement the migration of other kind of activities
111       simgrid::kernel::activity::ExecImplPtr exec =
112           boost::dynamic_pointer_cast<simgrid::kernel::activity::ExecImpl>(pimpl_->waiting_synchro);
113       xbt_assert(exec.get() != nullptr, "We can only migrate blocked actors when they are blocked on executions.");
114       exec->migrate(new_host);
115     }
116     this->pimpl_->set_host(new_host);
117   });
118
119   s4u::Actor::on_migration_end(this);
120 }
121
122 s4u::Host* Actor::get_host()
123 {
124   return this->pimpl_->get_host();
125 }
126
127 void Actor::daemonize()
128 {
129   simgrid::simix::simcall([this]() { pimpl_->daemonize(); });
130 }
131
132 bool Actor::is_daemon() const
133 {
134   return this->pimpl_->is_daemon();
135 }
136
137 const simgrid::xbt::string& Actor::get_name() const
138 {
139   return this->pimpl_->get_name();
140 }
141
142 const char* Actor::get_cname() const
143 {
144   return this->pimpl_->get_cname();
145 }
146
147 aid_t Actor::get_pid() const
148 {
149   return this->pimpl_->get_pid();
150 }
151
152 aid_t Actor::get_ppid() const
153 {
154   return this->pimpl_->get_ppid();
155 }
156
157 void Actor::suspend()
158 {
159   s4u::Actor::on_suspend(this);
160   simcall_process_suspend(pimpl_);
161 }
162
163 void Actor::resume()
164 {
165   simgrid::simix::simcall([this] { pimpl_->resume(); });
166   s4u::Actor::on_resume(this);
167 }
168
169 bool Actor::is_suspended()
170 {
171   return simgrid::simix::simcall([this] { return pimpl_->is_suspended(); });
172 }
173
174 void Actor::set_kill_time(double kill_time)
175 {
176   simgrid::simix::simcall([this, kill_time] { pimpl_->set_kill_time(kill_time); });
177 }
178
179 /** @brief Get the kill time of an actor(or 0 if unset). */
180 double Actor::get_kill_time()
181 {
182   return pimpl_->get_kill_time();
183 }
184
185 void Actor::kill(aid_t pid) // deprecated
186 {
187   smx_actor_t killer  = SIMIX_process_self();
188   smx_actor_t victim  = SIMIX_process_from_PID(pid);
189   if (victim != nullptr) {
190     simgrid::simix::simcall([killer, victim] { killer->kill(victim); });
191   } else {
192     std::ostringstream oss;
193     oss << "kill: (" << pid << ") - No such actor" << std::endl;
194     throw std::runtime_error(oss.str());
195   }
196 }
197
198 void Actor::kill()
199 {
200   smx_actor_t process = SIMIX_process_self();
201   simgrid::simix::simcall([this, process] {
202     if (pimpl_ == simix_global->maestro_process)
203       pimpl_->exit();
204     else
205       process->kill(pimpl_);
206   });
207 }
208
209 smx_actor_t Actor::get_impl()
210 {
211   return pimpl_;
212 }
213
214 // ***** Static functions *****
215
216 ActorPtr Actor::by_pid(aid_t pid)
217 {
218   smx_actor_t process = SIMIX_process_from_PID(pid);
219   if (process != nullptr)
220     return process->iface();
221   else
222     return ActorPtr();
223 }
224
225 void Actor::kill_all()
226 {
227   smx_actor_t self = SIMIX_process_self();
228   simgrid::simix::simcall([self] { self->kill_all(); });
229 }
230
231 std::unordered_map<std::string, std::string>* Actor::get_properties()
232 {
233   return simgrid::simix::simcall([this] { return this->pimpl_->get_properties(); });
234 }
235
236 /** Retrieve the property value (or nullptr if not set) */
237 const char* Actor::get_property(const std::string& key)
238 {
239   return simgrid::simix::simcall([this, key] { return pimpl_->get_property(key); });
240 }
241
242 void Actor::set_property(const std::string& key, std::string value)
243 {
244   simgrid::simix::simcall([this, key, value] { pimpl_->set_property(key, std::move(value)); });
245 }
246
247 Actor* Actor::restart()
248 {
249   return simgrid::simix::simcall([this]() { return pimpl_->restart(); });
250 }
251
252 // ***** this_actor *****
253
254 namespace this_actor {
255
256 /** Returns true if run from the kernel mode, and false if run from a real actor
257  *
258  * Everything that is run out of any actor (simulation setup before the engine is run,
259  * computing the model evolutions as a result to the actors' action, etc) is run in
260  * kernel mode, just as in any operating systems.
261  *
262  * In SimGrid, the actor in charge of doing the stuff in kernel mode is called Maestro,
263  * because it is the one scheduling when the others should move or wait.
264  */
265 bool is_maestro()
266 {
267   smx_actor_t process = SIMIX_process_self();
268   return process == nullptr || process == simix_global->maestro_process;
269 }
270
271 void sleep_for(double duration)
272 {
273   if (duration > 0) {
274     smx_actor_t actor = SIMIX_process_self();
275     simgrid::s4u::Actor::on_sleep(actor->iface());
276
277     simcall_process_sleep(duration);
278
279     simgrid::s4u::Actor::on_wake_up(actor->iface());
280   }
281 }
282
283 void yield()
284 {
285   simgrid::simix::simcall([] { /* do nothing*/ });
286 }
287
288 XBT_PUBLIC void sleep_until(double timeout)
289 {
290   double now = SIMIX_get_clock();
291   if (timeout > now)
292     sleep_for(timeout - now);
293 }
294
295 void execute(double flops)
296 {
297   execute(flops, 1.0 /* priority */);
298 }
299
300 void execute(double flops, double priority)
301 {
302   exec_init(flops)->set_priority(priority)->start()->wait();
303 }
304
305 void parallel_execute(const std::vector<s4u::Host*>& hosts, const std::vector<double>& flops_amounts,
306                       const std::vector<double>& bytes_amounts)
307 {
308   parallel_execute(hosts, flops_amounts, bytes_amounts, -1);
309 }
310
311 void parallel_execute(const std::vector<s4u::Host*>& hosts, const std::vector<double>& flops_amounts,
312                       const std::vector<double>& bytes_amounts, double timeout)
313 {
314   xbt_assert(hosts.size() > 0, "Your parallel executions must span over at least one host.");
315   xbt_assert(hosts.size() == flops_amounts.size() || flops_amounts.empty(),
316              "Host count (%zu) does not match flops_amount count (%zu).", hosts.size(), flops_amounts.size());
317   xbt_assert(hosts.size() * hosts.size() == bytes_amounts.size() || bytes_amounts.empty(),
318              "bytes_amounts must be a matrix of size host_count * host_count (%zu*%zu), but it's of size %zu.",
319              hosts.size(), hosts.size(), flops_amounts.size());
320   /* Check that we are not mixing VMs and PMs in the parallel task */
321   bool is_a_vm = (nullptr != dynamic_cast<simgrid::s4u::VirtualMachine*>(hosts.front()));
322   xbt_assert(std::all_of(hosts.begin(), hosts.end(),
323                          [is_a_vm](s4u::Host* elm) {
324                            bool tmp_is_a_vm = (nullptr != dynamic_cast<simgrid::s4u::VirtualMachine*>(elm));
325                            return is_a_vm == tmp_is_a_vm;
326                          }),
327              "parallel_execute: mixing VMs and PMs is not supported (yet).");
328   /* checking for infinite values */
329   xbt_assert(std::all_of(flops_amounts.begin(), flops_amounts.end(), [](double elm) { return std::isfinite(elm); }),
330              "flops_amounts comprises infinite values!");
331   xbt_assert(std::all_of(bytes_amounts.begin(), bytes_amounts.end(), [](double elm) { return std::isfinite(elm); }),
332              "flops_amounts comprises infinite values!");
333
334   exec_init(hosts, flops_amounts, bytes_amounts)->set_timeout(timeout)->wait();
335 }
336
337 // deprecated
338 void parallel_execute(int host_nb, s4u::Host* const* host_list, const double* flops_amount, const double* bytes_amount,
339                       double timeout)
340 {
341   smx_activity_t s =
342       simcall_execution_parallel_start("", host_nb, host_list, flops_amount, bytes_amount, /* rate */ -1, timeout);
343   simcall_execution_wait(s);
344   delete[] flops_amount;
345   delete[] bytes_amount;
346 }
347
348 // deprecated
349 void parallel_execute(int host_nb, s4u::Host* const* host_list, const double* flops_amount, const double* bytes_amount)
350 {
351   smx_activity_t s = simcall_execution_parallel_start("", host_nb, host_list, flops_amount, bytes_amount,
352                                                       /* rate */ -1, /*timeout*/ -1);
353   simcall_execution_wait(s);
354   delete[] flops_amount;
355   delete[] bytes_amount;
356 }
357
358 ExecPtr exec_init(double flops_amount)
359 {
360   return ExecPtr(new ExecSeq(get_host(), flops_amount));
361 }
362
363 ExecPtr exec_init(const std::vector<s4u::Host*>& hosts, const std::vector<double>& flops_amounts,
364                   const std::vector<double>& bytes_amounts)
365 {
366   return ExecPtr(new ExecPar(hosts, flops_amounts, bytes_amounts));
367 }
368
369 ExecPtr exec_async(double flops)
370 {
371   ExecPtr res = exec_init(flops);
372   res->start();
373   return res;
374 }
375
376 aid_t get_pid()
377 {
378   return SIMIX_process_self()->get_pid();
379 }
380
381 aid_t get_ppid()
382 {
383   return SIMIX_process_self()->get_ppid();
384 }
385
386 std::string get_name()
387 {
388   return SIMIX_process_self()->get_name();
389 }
390
391 const char* get_cname()
392 {
393   return SIMIX_process_self()->get_cname();
394 }
395
396 Host* get_host()
397 {
398   return SIMIX_process_self()->get_host();
399 }
400
401 void suspend()
402 {
403   smx_actor_t actor = SIMIX_process_self();
404   simgrid::s4u::Actor::on_suspend(actor->iface());
405
406   simcall_process_suspend(actor);
407 }
408
409 void resume()
410 {
411   smx_actor_t process = SIMIX_process_self();
412   simgrid::simix::simcall([process] { process->resume(); });
413   simgrid::s4u::Actor::on_resume(process->iface());
414 }
415
416 void exit()
417 {
418   smx_actor_t actor = SIMIX_process_self();
419   simgrid::simix::simcall([actor] { actor->exit(); });
420 }
421
422 void on_exit(std::function<void(bool)> const fun)
423 {
424   SIMIX_process_self()->iface()->on_exit(fun);
425 }
426
427 void on_exit(std::function<void(int, void*)> const fun, void* data) /* deprecated */
428 {
429   SIMIX_process_self()->iface()->on_exit([fun, data](bool exit) { fun(exit, data); });
430 }
431
432 /** @brief Moves the current actor to another host
433  *
434  * @see simgrid::s4u::Actor::migrate() for more information
435  */
436 void migrate(Host* new_host)
437 {
438   SIMIX_process_self()->iface()->migrate(new_host);
439 }
440
441 std::string getName() /* deprecated */
442 {
443   return get_name();
444 }
445 const char* getCname() /* deprecated */
446 {
447   return get_cname();
448 }
449 bool isMaestro() /* deprecated */
450 {
451   return is_maestro();
452 }
453 aid_t getPid() /* deprecated */
454 {
455   return get_pid();
456 }
457 aid_t getPpid() /* deprecated */
458 {
459   return get_ppid();
460 }
461 Host* getHost() /* deprecated */
462 {
463   return get_host();
464 }
465 void on_exit(int_f_pvoid_pvoid_t fun, void* data) /* deprecated */
466 {
467   SIMIX_process_self()->iface()->on_exit([fun, data](int a) { fun((void*)(intptr_t)a, data); });
468 }
469 void onExit(int_f_pvoid_pvoid_t fun, void* data) /* deprecated */
470 {
471   on_exit([fun, data](int a) { fun((void*)(intptr_t)a, data); });
472 }
473 void kill() /* deprecated */
474 {
475   exit();
476 }
477
478 } // namespace this_actor
479 } // namespace s4u
480 } // namespace simgrid
481
482 /* **************************** Public C interface *************************** */
483
484 /** @ingroup m_actor_management
485  * @brief Returns the process ID of @a actor.
486  *
487  * This function checks whether @a actor is a valid pointer and return its PID (or 0 in case of problem).
488  */
489 aid_t sg_actor_get_PID(sg_actor_t actor)
490 {
491   /* Do not raise an exception here: this function is called by the logs
492    * and the exceptions, so it would be called back again and again */
493   if (actor == nullptr || actor->get_impl() == nullptr)
494     return 0;
495   return actor->get_pid();
496 }
497
498 /** @ingroup m_actor_management
499  * @brief Returns the process ID of the parent of @a actor.
500  *
501  * This function checks whether @a actor is a valid pointer and return its parent's PID.
502  * Returns -1 if the actor has not been created by any other actor.
503  */
504 aid_t sg_actor_get_PPID(sg_actor_t actor)
505 {
506   return actor->get_ppid();
507 }
508
509 /** @ingroup m_actor_management
510  *
511  * @brief Return a #sg_actor_t given its PID.
512  *
513  * This function search in the list of all the created sg_actor_t for a sg_actor_t  whose PID is equal to @a PID.
514  * If none is found, @c nullptr is returned.
515    Note that the PID are unique in the whole simulation, not only on a given host.
516  */
517 sg_actor_t sg_actor_by_PID(aid_t pid)
518 {
519   return simgrid::s4u::Actor::by_pid(pid).get();
520 }
521
522 /** @ingroup m_actor_management
523  * @brief Return the name of an actor.
524  */
525 const char* sg_actor_get_name(sg_actor_t actor)
526 {
527   return actor->get_cname();
528 }
529
530 sg_host_t sg_actor_get_host(sg_actor_t actor)
531 {
532   return actor->get_host();
533 }
534
535 /** @ingroup m_actor_management
536  * @brief Returns the value of a given actor property
537  *
538  * @param actor an actor
539  * @param name a property name
540  * @return value of a property (or nullptr if the property is not set)
541  */
542 const char* sg_actor_get_property_value(sg_actor_t actor, const char* name)
543 {
544   return actor->get_property(name);
545 }
546
547 /** @ingroup m_actor_management
548  * @brief Return the list of properties
549  *
550  * This function returns all the parameters associated with an actor
551  */
552 xbt_dict_t sg_actor_get_properties(sg_actor_t actor)
553 {
554   xbt_assert(actor != nullptr, "Invalid parameter: First argument must not be nullptr");
555   xbt_dict_t as_dict                        = xbt_dict_new_homogeneous(xbt_free_f);
556   std::unordered_map<std::string, std::string>* props = actor->get_properties();
557   if (props == nullptr)
558     return nullptr;
559   for (auto const& kv : *props) {
560     xbt_dict_set(as_dict, kv.first.c_str(), xbt_strdup(kv.second.c_str()), nullptr);
561   }
562   return as_dict;
563 }
564
565 /** @ingroup m_actor_management
566  * @brief Suspend the actor.
567  *
568  * This function suspends the actor by suspending the task on which it was waiting for the completion.
569  */
570 void sg_actor_suspend(sg_actor_t actor)
571 {
572   xbt_assert(actor != nullptr, "Invalid parameter: First argument must not be nullptr");
573   actor->suspend();
574 }
575
576 /** @ingroup m_actor_management
577  * @brief Resume a suspended actor.
578  *
579  * This function resumes a suspended actor by resuming the task on which it was waiting for the completion.
580  */
581 void sg_actor_resume(sg_actor_t actor)
582 {
583   xbt_assert(actor != nullptr, "Invalid parameter: First argument must not be nullptr");
584   actor->resume();
585 }
586
587 /** @ingroup m_actor_management
588  * @brief Returns true if the actor is suspended .
589  *
590  * This checks whether an actor is suspended or not by inspecting the task on which it was waiting for the completion.
591  */
592 int sg_actor_is_suspended(sg_actor_t actor)
593 {
594   return actor->is_suspended();
595 }
596
597 /**
598  * @ingroup m_actor_management
599  * @brief Restarts an actor from the beginning.
600  */
601 sg_actor_t sg_actor_restart(sg_actor_t actor)
602 {
603   return actor->restart();
604 }
605
606 /**
607  * @ingroup m_actor_management
608  * @brief Sets the "auto-restart" flag of the actor.
609  * If the flag is set to 1, the actor will be automatically restarted when its host comes back up.
610  */
611 void sg_actor_set_auto_restart(sg_actor_t actor, int auto_restart)
612 {
613   actor->set_auto_restart(auto_restart);
614 }
615
616 /** @ingroup m_actor_management
617  * @brief This actor will be terminated automatically when the last non-daemon actor finishes
618  */
619 void sg_actor_daemonize(sg_actor_t actor)
620 {
621   actor->daemonize();
622 }
623
624 /** @ingroup m_actor_management
625  * @brief Migrates an actor to another location.
626  *
627  * This function changes the value of the #sg_host_t on  which @a actor is running.
628  */
629 void sg_actor_migrate(sg_actor_t process, sg_host_t host)
630 {
631   process->migrate(host);
632 }
633
634 /** @ingroup m_actor_management
635  * @brief Wait for the completion of a #sg_actor_t.
636  *
637  * @param actor the actor to wait for
638  * @param timeout wait until the actor is over, or the timeout expires
639  */
640 void sg_actor_join(sg_actor_t actor, double timeout)
641 {
642   actor->join(timeout);
643 }
644
645 void sg_actor_kill(sg_actor_t actor)
646 {
647   actor->kill();
648 }
649
650 void sg_actor_kill_all()
651 {
652   simgrid::s4u::Actor::kill_all();
653 }
654
655 /** @ingroup m_actor_management
656  * @brief Set the kill time of an actor.
657  *
658  * @param actor an actor
659  * @param kill_time the time when the actor is killed.
660  */
661 void sg_actor_set_kill_time(sg_actor_t actor, double kill_time)
662 {
663   actor->set_kill_time(kill_time);
664 }
665
666 /** Yield the current actor; let the other actors execute first */
667 void sg_actor_yield()
668 {
669   simgrid::s4u::this_actor::yield();
670 }
671
672 void sg_actor_sleep_for(double duration)
673 {
674   simgrid::s4u::this_actor::sleep_for(duration);
675 }
676
677 sg_actor_t sg_actor_attach(const char* name, void* data, sg_host_t host, xbt_dict_t properties)
678 {
679   xbt_assert(host != nullptr, "Invalid parameters: host and code params must not be nullptr");
680   std::unordered_map<std::string, std::string> props;
681   xbt_dict_cursor_t cursor = nullptr;
682   char* key;
683   char* value;
684   xbt_dict_foreach (properties, cursor, key, value)
685     props[key] = value;
686   xbt_dict_free(&properties);
687
688   /* Let's create the process: SIMIX may decide to start it right now, even before returning the flow control to us */
689   smx_actor_t actor = nullptr;
690   try {
691     actor = simgrid::kernel::actor::ActorImpl::attach(name, data, host, &props).get();
692   } catch (simgrid::HostFailureException const&) {
693     xbt_die("Could not attach");
694   }
695
696   simgrid::s4u::this_actor::yield();
697   return actor->ciface();
698 }
699
700 void sg_actor_detach()
701 {
702   simgrid::kernel::actor::ActorImpl::detach();
703 }