From 9b509521ea81de97a210ca785ad29b3be310153f Mon Sep 17 00:00:00 2001 From: Frederic Suter Date: Thu, 6 Jul 2017 11:24:50 +0200 Subject: [PATCH] make sleep and suspend methods of ActorImpl --- src/plugins/vm/VirtualMachineImpl.cpp | 2 +- src/simix/ActorImpl.cpp | 81 +++++++++++++-------------- src/simix/ActorImpl.hpp | 6 +- 3 files changed, 43 insertions(+), 46 deletions(-) diff --git a/src/plugins/vm/VirtualMachineImpl.cpp b/src/plugins/vm/VirtualMachineImpl.cpp index 304a07f0d5..019fd79f73 100644 --- a/src/plugins/vm/VirtualMachineImpl.cpp +++ b/src/plugins/vm/VirtualMachineImpl.cpp @@ -174,7 +174,7 @@ void VirtualMachineImpl::suspend(smx_actor_t issuer) smx_actor_t smx_process_safe; xbt_swag_foreach_safe(smx_process, smx_process_safe, process_list) { XBT_DEBUG("suspend %s", smx_process->name.c_str()); - SIMIX_process_suspend(smx_process, issuer); + smx_process->suspend(issuer); } XBT_DEBUG("suspend all processes on the VM done done"); diff --git a/src/simix/ActorImpl.cpp b/src/simix/ActorImpl.cpp index 79356f08b6..36d847fb8d 100644 --- a/src/simix/ActorImpl.cpp +++ b/src/simix/ActorImpl.cpp @@ -167,6 +167,27 @@ void ActorImpl::daemonize() } } +smx_activity_t ActorImpl::suspend(smx_actor_t issuer) +{ + if (suspended) { + XBT_DEBUG("Process '%s' is already suspended", name.c_str()); + return nullptr; + } + + suspended = 1; + + /* If we are suspending another process that is waiting on a sync, suspend its synchronization. */ + if (this != issuer) { + if (waiting_synchro) + waiting_synchro->suspend(); + /* If the other process is not waiting, its suspension is delayed to when the process is rescheduled. */ + + return nullptr; + } else { + return SIMIX_execution_start(this, "suspend", 0.0, 1.0, 0.0); + } +} + void ActorImpl::resume() { XBT_IN("process = %p", this); @@ -187,6 +208,20 @@ void ActorImpl::resume() XBT_OUT(); } +smx_activity_t ActorImpl::sleep(double duration) +{ + if (host->isOff()) + THROWF(host_error, 0, "Host %s failed, you cannot sleep there.", host->getCname()); + + simgrid::kernel::activity::SleepImpl* synchro = new simgrid::kernel::activity::SleepImpl(); + synchro->host = host; + synchro->surf_sleep = host->pimpl_cpu->sleep(duration); + synchro->surf_sleep->setData(synchro); + XBT_DEBUG("Create sleep synchronization %p", synchro); + + return synchro; +} + void create_maestro(std::function code) { smx_actor_t maestro = nullptr; @@ -553,7 +588,7 @@ void SIMIX_process_change_host(smx_actor_t process, sg_host_t dest) void simcall_HANDLER_process_suspend(smx_simcall_t simcall, smx_actor_t process) { - smx_activity_t sync_suspend = SIMIX_process_suspend(process, simcall->issuer); + smx_activity_t sync_suspend = process->suspend(simcall->issuer); if (process != simcall->issuer) { SIMIX_simcall_answer(simcall); @@ -565,28 +600,6 @@ void simcall_HANDLER_process_suspend(smx_simcall_t simcall, smx_actor_t process) /* If we are suspending ourselves, then just do not finish the simcall now */ } -smx_activity_t SIMIX_process_suspend(smx_actor_t process, smx_actor_t issuer) -{ - if (process->suspended) { - XBT_DEBUG("Process '%s' is already suspended", process->name.c_str()); - return nullptr; - } - - process->suspended = 1; - - /* If we are suspending another process that is waiting on a sync, suspend its synchronization. */ - if (process != issuer) { - - if (process->waiting_synchro) - process->waiting_synchro->suspend(); - /* If the other process is not waiting, its suspension is delayed to when the process is rescheduled. */ - - return nullptr; - } else { - return SIMIX_execution_start(process, "suspend", 0.0, 1.0, 0.0); - } -} - int SIMIX_process_get_maxpid() { return simix_process_maxpid; } @@ -679,7 +692,7 @@ static int SIMIX_process_join_finish(smx_process_exit_status_t status, void* syn smx_activity_t SIMIX_process_join(smx_actor_t issuer, smx_actor_t process, double timeout) { - smx_activity_t res = SIMIX_process_sleep(issuer, timeout); + smx_activity_t res = issuer->sleep(timeout); intrusive_ptr_add_ref(res.get()); /* We are leaking the process here, but if we don't take the ref, we get a "use after free". * The correct solution would be to derivate the type SynchroSleep into a SynchroProcessJoin, @@ -699,27 +712,11 @@ void simcall_HANDLER_process_sleep(smx_simcall_t simcall, double duration) SIMIX_simcall_answer(simcall); return; } - smx_activity_t sync = SIMIX_process_sleep(simcall->issuer, duration); + smx_activity_t sync = simcall->issuer->sleep(duration); sync->simcalls.push_back(simcall); simcall->issuer->waiting_synchro = sync; } -smx_activity_t SIMIX_process_sleep(smx_actor_t process, double duration) -{ - sg_host_t host = process->host; - - if (host->isOff()) - THROWF(host_error, 0, "Host %s failed, you cannot sleep there.", host->getCname()); - - simgrid::kernel::activity::SleepImpl* synchro = new simgrid::kernel::activity::SleepImpl(); - synchro->host = host; - synchro->surf_sleep = host->pimpl_cpu->sleep(duration); - synchro->surf_sleep->setData(synchro); - XBT_DEBUG("Create sleep synchronization %p", synchro); - - return synchro; -} - void SIMIX_process_sleep_destroy(smx_activity_t synchro) { XBT_DEBUG("Destroy sleep synchro %p", synchro.get()); @@ -776,7 +773,7 @@ void SIMIX_process_yield(smx_actor_t self) XBT_DEBUG("Hey! I'm suspended."); xbt_assert(self->exception != nullptr, "Gasp! This exception may be lost by subsequent calls."); self->suspended = 0; - SIMIX_process_suspend(self, self); + self->suspend(self); } if (self->exception != nullptr) { diff --git a/src/simix/ActorImpl.hpp b/src/simix/ActorImpl.hpp index e7dd3b0d06..ba816b2a0d 100644 --- a/src/simix/ActorImpl.hpp +++ b/src/simix/ActorImpl.hpp @@ -56,7 +56,7 @@ public: bool suspended = false; bool auto_restart = false; - sg_host_t new_host = nullptr; /* if not null, the host on which the process must migrate to */ + sg_host_t new_host = nullptr; /* if not null, the host on which the process must migrate to */ smx_activity_t waiting_synchro = nullptr; /* the current blocking synchro if any */ std::list comms; /* the current non-blocking communication synchros */ xbt_dict_t properties = nullptr; @@ -106,7 +106,9 @@ public: void daemonize(); bool isDaemon() { return daemon; } /** Whether this actor has been daemonized */ bool isSuspended() { return suspended; } + smx_activity_t suspend(smx_actor_t issuer); void resume(); + smx_activity_t sleep(double duration); }; } @@ -145,8 +147,6 @@ extern void (*SMPI_switch_data_segment)(int dest); SG_END_DECL() XBT_PRIVATE void SIMIX_process_sleep_destroy(smx_activity_t synchro); -XBT_PRIVATE smx_activity_t SIMIX_process_suspend(smx_actor_t process, smx_actor_t issuer); XBT_PRIVATE smx_activity_t SIMIX_process_join(smx_actor_t issuer, smx_actor_t process, double timeout); -XBT_PRIVATE smx_activity_t SIMIX_process_sleep(smx_actor_t process, double duration); #endif -- 2.20.1