X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/b1020c2311048f26a586af241101434faa59445b..09392faf42646e631a4e42553e901410eb3e488e:/src/s4u/s4u_Activity.cpp diff --git a/src/s4u/s4u_Activity.cpp b/src/s4u/s4u_Activity.cpp index 529d08f471..a1711b5fb5 100644 --- a/src/s4u/s4u_Activity.cpp +++ b/src/s4u/s4u_Activity.cpp @@ -1,20 +1,43 @@ -/* Copyright (c) 2006-2021. The SimGrid Team. All rights reserved. */ +/* Copyright (c) 2006-2023. The SimGrid Team. All rights reserved. */ /* 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 "xbt/log.h" +#include +#include +#include +#include +#include +#include +#include +#include -#include "simgrid/s4u/Activity.hpp" -#include "simgrid/s4u/Engine.hpp" #include "src/kernel/activity/ActivityImpl.hpp" +#include "src/kernel/actor/ActorImpl.hpp" +#include "src/kernel/actor/CommObserver.hpp" XBT_LOG_EXTERNAL_CATEGORY(s4u); XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_activity, s4u, "S4U activities"); namespace simgrid { + +template class xbt::Extendable; + namespace s4u { +std::set* Activity::vetoed_activities_ = nullptr; + +void Activity::destroy() +{ + /* First Remove all dependencies */ + while (not dependencies_.empty()) + (*(dependencies_.begin()))->remove_successor(this); + while (not successors_.empty()) + this->remove_successor(successors_.front()); + + cancel(); +} + void Activity::wait_until(double time_limit) { double now = Engine::get_clock(); @@ -22,6 +45,32 @@ void Activity::wait_until(double time_limit) wait_for(time_limit - now); } +Activity* Activity::wait_for(double timeout) +{ + if (state_ == State::INITED) + start(); + + if (state_ == State::FAILED) { + if (dynamic_cast(this)) + throw NetworkFailureException(XBT_THROW_POINT, "Cannot wait for a failed comm"); + if (dynamic_cast(this)) + throw NetworkFailureException(XBT_THROW_POINT, "Cannot wait for a failed mess"); + if (dynamic_cast(this)) + throw HostFailureException(XBT_THROW_POINT, "Cannot wait for a failed exec"); + if (dynamic_cast(this)) + throw StorageFailureException(XBT_THROW_POINT, "Cannot wait for a failed I/O"); + THROW_IMPOSSIBLE; + } + + kernel::actor::ActorImpl* issuer = kernel::actor::ActorImpl::self(); + kernel::actor::ActivityWaitSimcall observer{issuer, pimpl_.get(), timeout, "wait_for"}; + if (kernel::actor::simcall_blocking( + [&observer] { observer.get_activity()->wait_for(observer.get_issuer(), observer.get_timeout()); }, &observer)) + throw TimeoutException(XBT_THROW_POINT, "Timeouted"); + complete(State::FINISHED); + return this; +} + bool Activity::test() { xbt_assert(state_ == State::INITED || state_ == State::STARTED || state_ == State::STARTING || @@ -31,17 +80,66 @@ bool Activity::test() return true; if (state_ == State::INITED || state_ == State::STARTING) - this->vetoable_start(); + this->start(); - if (kernel::actor::simcall([this] { return this->get_impl()->test(); })) { - state_ = State::FINISHED; - this->release_dependencies(); + kernel::actor::ActorImpl* issuer = kernel::actor::ActorImpl::self(); + kernel::actor::ActivityTestSimcall observer{issuer, pimpl_.get(), "test"}; + if (kernel::actor::simcall_answered([&observer] { return observer.get_activity()->test(observer.get_issuer()); }, + &observer)) { + complete(State::FINISHED); return true; } - return false; } +ssize_t Activity::test_any(const std::vector& activities) // XBT_ATTRIB_DEPRECATED_v339 +{ + std::vector ractivities(activities.size()); + std::transform(begin(activities), end(activities), begin(ractivities), + [](const ActivityPtr& act) { return act->pimpl_.get(); }); + + kernel::actor::ActorImpl* issuer = kernel::actor::ActorImpl::self(); + kernel::actor::ActivityTestanySimcall observer{issuer, ractivities, "test_any"}; + ssize_t changed_pos = kernel::actor::simcall_answered( + [&observer] { + return kernel::activity::ActivityImpl::test_any(observer.get_issuer(), observer.get_activities()); + }, + &observer); + if (changed_pos != -1) + activities.at(changed_pos)->complete(State::FINISHED); + return changed_pos; +} + +ssize_t Activity::deprecated_wait_any_for(const std::vector& activities, double timeout) // XBT_ATTRIB_DEPRECATED_v339 +{ + std::vector ractivities(activities.size()); + std::transform(begin(activities), end(activities), begin(ractivities), + [](const ActivityPtr& activity) { return activity->pimpl_.get(); }); + + kernel::actor::ActorImpl* issuer = kernel::actor::ActorImpl::self(); + kernel::actor::ActivityWaitanySimcall observer{issuer, ractivities, timeout, "wait_any_for"}; + ssize_t changed_pos = kernel::actor::simcall_blocking( + [&observer] { + kernel::activity::ActivityImpl::wait_any_for(observer.get_issuer(), observer.get_activities(), + observer.get_timeout()); + }, + &observer); + if (changed_pos != -1) + activities.at(changed_pos)->complete(State::FINISHED); + return changed_pos; +} + +Activity* Activity::cancel() +{ + kernel::actor::simcall_answered([this] { + XBT_HERE(); + if (pimpl_) + pimpl_->cancel(); + }); + complete(State::CANCELED); + return this; +} + Activity* Activity::suspend() { if (suspended_) @@ -65,25 +163,9 @@ Activity* Activity::resume() return this; } -const char* Activity::get_state_str() +const char* Activity::get_state_str() const { - switch (state_) { - case State::INITED: - return "INITED"; - - case State::STARTING: - return "STARTING"; - - case State::STARTED: - return "STARTED"; - - case State::CANCELED: - return "CANCELED"; - - case State::FINISHED: - return "FINISHED"; - } - THROW_IMPOSSIBLE; + return to_c_str(state_); } double Activity::get_remaining() const @@ -93,10 +175,19 @@ double Activity::get_remaining() const else return pimpl_->get_remaining(); } +double Activity::get_start_time() const +{ + return pimpl_->get_start_time(); +} +double Activity::get_finish_time() const +{ + return pimpl_->get_finish_time(); +} Activity* Activity::set_remaining(double remains) { - xbt_assert(state_ == State::INITED, "Cannot change the remaining amount of work once the Activity is started"); + xbt_assert(state_ == State::INITED || state_ == State::STARTING, + "Cannot change the remaining amount of work once the Activity is started"); remains_ = remains; return this; }