X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/29a3b2869c0075fc75e8ccc66fc1d9c4c8bf6a85..b372d23d3d8818b4547ef207f43b45709298a002:/src/s4u/s4u_Comm.cpp diff --git a/src/s4u/s4u_Comm.cpp b/src/s4u/s4u_Comm.cpp index 658097ea50..55a79e5aa7 100644 --- a/src/s4u/s4u_Comm.cpp +++ b/src/s4u/s4u_Comm.cpp @@ -1,4 +1,4 @@ -/* Copyright (c) 2006-2018. The SimGrid Team. All rights reserved. */ +/* Copyright (c) 2006-2020. 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. */ @@ -6,50 +6,44 @@ #include "src/msg/msg_private.hpp" #include "xbt/log.h" +#include "simgrid/Exception.hpp" #include "simgrid/s4u/Comm.hpp" #include "simgrid/s4u/Mailbox.hpp" +#include + XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_comm, s4u_activity, "S4U asynchronous communications"); namespace simgrid { namespace s4u { -simgrid::xbt::signal s4u::Comm::on_sender_start; -simgrid::xbt::signal s4u::Comm::on_receiver_start; -simgrid::xbt::signal s4u::Comm::on_completion; +xbt::signal Comm::on_start; +xbt::signal Comm::on_completion; Comm::~Comm() { - if (state_ == State::STARTED && not detached_ && (pimpl_ == nullptr || pimpl_->state_ == SIMIX_RUNNING)) { + if (state_ == State::STARTED && not detached_ && + (pimpl_ == nullptr || pimpl_->state_ == kernel::activity::State::RUNNING)) { XBT_INFO("Comm %p freed before its completion. Detached: %d, State: %d", this, detached_, (int)state_); if (pimpl_ != nullptr) - XBT_INFO("pimpl_->state: %d", pimpl_->state_); + XBT_INFO("pimpl_->state: %d", static_cast(pimpl_->state_)); else XBT_INFO("pimpl_ is null"); xbt_backtrace_display_current(); } } -int Comm::wait_any_for(std::vector* comms_in, double timeout) +int Comm::wait_any_for(const std::vector* comms, double timeout) { - // Map to dynar: - xbt_dynar_t comms = xbt_dynar_new(sizeof(simgrid::kernel::activity::ActivityImpl*), [](void* ptr) { - intrusive_ptr_release(*(simgrid::kernel::activity::ActivityImpl**)ptr); - }); - for (auto const& comm : *comms_in) { - if (comm->state_ == Activity::State::INITED) - comm->start(); - xbt_assert(comm->state_ == Activity::State::STARTED); - simgrid::kernel::activity::ActivityImpl* ptr = comm->pimpl_.get(); - intrusive_ptr_add_ref(ptr); - xbt_dynar_push_as(comms, simgrid::kernel::activity::ActivityImpl*, ptr); - } - // Call the underlying simcall: - int idx = simcall_comm_waitany(comms, timeout); - xbt_dynar_free(&comms); - return idx; + std::vector rcomms(comms->size()); + std::transform(begin(*comms), end(*comms), begin(rcomms), + [](const CommPtr& comm) { return static_cast(comm->pimpl_.get()); }); + int changed_pos = simcall_comm_waitany(rcomms.data(), rcomms.size(), timeout); + if (changed_pos != -1) + comms->at(changed_pos)->release_dependencies(); + return changed_pos; } -void Comm::wait_all(std::vector* comms) +void Comm::wait_all(const std::vector* comms) { // TODO: this should be a simcall or something // TODO: we are missing a version with timeout @@ -57,7 +51,7 @@ void Comm::wait_all(std::vector* comms) comm->wait(); } -Comm* Comm::set_rate(double rate) +CommPtr Comm::set_rate(double rate) { xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", __FUNCTION__); @@ -65,7 +59,7 @@ Comm* Comm::set_rate(double rate) return this; } -Comm* Comm::set_src_data(void* buff) +CommPtr Comm::set_src_data(void* buff) { xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", __FUNCTION__); @@ -73,14 +67,16 @@ Comm* Comm::set_src_data(void* buff) src_buff_ = buff; return this; } -Comm* Comm::set_src_data_size(size_t size) + +CommPtr Comm::set_src_data_size(size_t size) { xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", __FUNCTION__); src_buff_size_ = size; return this; } -Comm* Comm::set_src_data(void* buff, size_t size) + +CommPtr Comm::set_src_data(void* buff, size_t size) { xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", __FUNCTION__); @@ -90,7 +86,8 @@ Comm* Comm::set_src_data(void* buff, size_t size) src_buff_size_ = size; return this; } -Comm* Comm::set_dst_data(void** buff) + +CommPtr Comm::set_dst_data(void** buff) { xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", __FUNCTION__); @@ -98,12 +95,13 @@ Comm* Comm::set_dst_data(void** buff) dst_buff_ = buff; return this; } -size_t Comm::get_dst_data_size() + +size_t Comm::get_dst_data_size() const { xbt_assert(state_ == State::FINISHED, "You cannot use %s before your communication terminated", __FUNCTION__); return dst_buff_size_; } -Comm* Comm::set_dst_data(void** buff, size_t size) +CommPtr Comm::set_dst_data(void** buff, size_t size) { xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", __FUNCTION__); @@ -116,22 +114,26 @@ Comm* Comm::set_dst_data(void** buff, size_t size) Comm* Comm::start() { - xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", - __FUNCTION__); + xbt_assert(get_state() == State::INITED || get_state() == State::STARTING, + "You cannot use %s() once your communication started (not implemented)", __FUNCTION__); if (src_buff_ != nullptr) { // Sender side - on_sender_start(Actor::self()); + on_start(*this, true /* is_sender*/); pimpl_ = simcall_comm_isend(sender_, mailbox_->get_impl(), remains_, rate_, src_buff_, src_buff_size_, match_fun_, - clean_fun_, copy_data_function_, user_data_, detached_); + clean_fun_, copy_data_function_, get_user_data(), detached_); } else if (dst_buff_ != nullptr) { // Receiver side xbt_assert(not detached_, "Receive cannot be detached"); - on_receiver_start(Actor::self()); + on_start(*this, false /*is_sender*/); pimpl_ = simcall_comm_irecv(receiver_, mailbox_->get_impl(), dst_buff_, &dst_buff_size_, match_fun_, - copy_data_function_, user_data_, rate_); + copy_data_function_, get_user_data(), rate_); } else { xbt_die("Cannot start a communication before specifying whether we are the sender or the receiver"); } + + if (suspended_) + pimpl_->suspend(); + state_ = State::STARTED; return this; } @@ -144,7 +146,7 @@ Comm* Comm::wait() /** @brief Block the calling actor until the communication is finished, or until timeout * - * On timeout, an exception is thrown. + * On timeout, an exception is thrown and the communication is invalidated. * * @param timeout the amount of seconds to wait for the comm termination. * Negative values denote infinite wait times. 0 as a timeout returns immediately. */ @@ -152,42 +154,49 @@ Comm* Comm::wait_for(double timeout) { switch (state_) { case State::FINISHED: - return this; + break; - case State::INITED: // It's not started yet. Do it in one simcall + case State::INITED: + case State::STARTING: // It's not started yet. Do it in one simcall if (src_buff_ != nullptr) { - on_sender_start(Actor::self()); + on_start(*this, true /*is_sender*/); simcall_comm_send(sender_, mailbox_->get_impl(), remains_, rate_, src_buff_, src_buff_size_, match_fun_, - copy_data_function_, user_data_, timeout); + copy_data_function_, get_user_data(), timeout); } else { // Receiver - on_receiver_start(Actor::self()); + on_start(*this, false /*is_sender*/); simcall_comm_recv(receiver_, mailbox_->get_impl(), dst_buff_, &dst_buff_size_, match_fun_, copy_data_function_, - user_data_, timeout, rate_); + get_user_data(), timeout, rate_); } state_ = State::FINISHED; - return this; + this->release_dependencies(); + break; case State::STARTED: - simcall_comm_wait(pimpl_, timeout); - on_completion(Actor::self()); + simcall_comm_wait(get_impl(), timeout); state_ = State::FINISHED; - return this; + this->release_dependencies(); + break; + + case State::CANCELED: + throw CancelException(XBT_THROW_POINT, "Communication canceled"); default: THROW_IMPOSSIBLE; } + on_completion(*this); return this; } -int Comm::test_any(std::vector* comms) + +int Comm::test_any(const std::vector* comms) { - smx_activity_t* array = new smx_activity_t[comms->size()]; - for (unsigned int i = 0; i < comms->size(); i++) { - array[i] = comms->at(i)->pimpl_; - } - int res = simcall_comm_testany(array, comms->size()); - delete[] array; - return res; + std::vector rcomms(comms->size()); + std::transform(begin(*comms), end(*comms), begin(rcomms), + [](const CommPtr& comm) { return static_cast(comm->pimpl_.get()); }); + int changed_pos = simcall_comm_testany(rcomms.data(), rcomms.size()); + if (changed_pos != -1) + comms->at(changed_pos)->release_dependencies(); + return changed_pos; } Comm* Comm::detach() @@ -196,48 +205,130 @@ Comm* Comm::detach() __FUNCTION__); xbt_assert(src_buff_ != nullptr && src_buff_size_ != 0, "You can only detach sends, not recvs"); detached_ = true; - return start(); + vetoable_start(); + return this; } Comm* Comm::cancel() { - simgrid::simix::simcall([this] { dynamic_cast(pimpl_.get())->cancel(); }); + kernel::actor::simcall([this] { + if (pimpl_) + boost::static_pointer_cast(pimpl_)->cancel(); + }); state_ = State::CANCELED; return this; } bool Comm::test() { - xbt_assert(state_ == State::INITED || state_ == State::STARTED || state_ == State::FINISHED); + xbt_assert(state_ == State::INITED || state_ == State::STARTED || state_ == State::STARTING || + state_ == State::FINISHED); if (state_ == State::FINISHED) return true; - if (state_ == State::INITED) - this->start(); + if (state_ == State::INITED || state_ == State::STARTING) + this->vetoable_start(); - if (simcall_comm_test(pimpl_)) { + if (simcall_comm_test(get_impl())) { state_ = State::FINISHED; + this->release_dependencies(); return true; } return false; } -MailboxPtr Comm::get_mailbox() +Mailbox* Comm::get_mailbox() const { return mailbox_; } -void intrusive_ptr_release(simgrid::s4u::Comm* c) +Actor* Comm::get_sender() const +{ + kernel::actor::ActorImplPtr sender = nullptr; + if (pimpl_) + sender = boost::static_pointer_cast(pimpl_)->src_actor_; + return sender ? sender->get_ciface() : nullptr; +} + +} // namespace s4u +} // namespace simgrid +/* **************************** Public C interface *************************** */ +void sg_comm_detach(sg_comm_t comm, void (*clean_function)(void*)) +{ + comm->detach(clean_function); + comm->unref(); +} +void sg_comm_unref(sg_comm_t comm) +{ + comm->unref(); +} +int sg_comm_test(sg_comm_t comm) +{ + bool finished = comm->test(); + if (finished) + comm->unref(); + return finished; +} + +sg_error_t sg_comm_wait(sg_comm_t comm) { - if (c->refcount_.fetch_sub(1, std::memory_order_release) == 1) { - std::atomic_thread_fence(std::memory_order_acquire); - delete c; + sg_error_t status = SG_OK; + + simgrid::s4u::CommPtr s4u_comm(comm, false); + try { + s4u_comm->wait_for(-1); + } catch (const simgrid::TimeoutException&) { + status = SG_ERROR_TIMEOUT; + } catch (const simgrid::CancelException&) { + status = SG_ERROR_CANCELED; + } catch (const simgrid::NetworkFailureException&) { + status = SG_ERROR_NETWORK; } + return status; } -void intrusive_ptr_add_ref(simgrid::s4u::Comm* c) + +sg_error_t sg_comm_wait_for(sg_comm_t comm, double timeout) { - c->refcount_.fetch_add(1, std::memory_order_relaxed); + sg_error_t status = SG_OK; + + simgrid::s4u::CommPtr s4u_comm(comm, false); + try { + s4u_comm->wait_for(timeout); + } catch (const simgrid::TimeoutException&) { + status = SG_ERROR_TIMEOUT; + } catch (const simgrid::CancelException&) { + status = SG_ERROR_CANCELED; + } catch (const simgrid::NetworkFailureException&) { + status = SG_ERROR_NETWORK; + } + return status; +} + +void sg_comm_wait_all(sg_comm_t* comms, size_t count) +{ + std::vector s4u_comms; + for (unsigned int i = 0; i < count; i++) + s4u_comms.emplace_back(comms[i], false); + + simgrid::s4u::Comm::wait_all(&s4u_comms); +} + +int sg_comm_wait_any(sg_comm_t* comms, size_t count) +{ + return sg_comm_wait_any_for(comms, count, -1); +} + +int sg_comm_wait_any_for(sg_comm_t* comms, size_t count, double timeout) +{ + std::vector s4u_comms; + for (unsigned int i = 0; i < count; i++) + s4u_comms.emplace_back(comms[i], false); + + int pos = simgrid::s4u::Comm::wait_any_for(&s4u_comms, timeout); + for (unsigned i = 0; i < count; i++) { + if (pos != -1 && static_cast(pos) != i) + s4u_comms[i]->add_ref(); + } + return pos; } -} // namespace s4u -} // namespace simgrid