X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/bba28202d80db522aef1af8dd81c9b9e3e0d3049..ac080087b39ef79ff497d3992ff04b3e20fe40b2:/src/s4u/s4u_Comm.cpp diff --git a/src/s4u/s4u_Comm.cpp b/src/s4u/s4u_Comm.cpp index c231d1e59e..98965dc78c 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-2019. 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,6 +6,7 @@ #include "src/msg/msg_private.hpp" #include "xbt/log.h" +#include "simgrid/Exception.hpp" #include "simgrid/s4u/Comm.hpp" #include "simgrid/s4u/Mailbox.hpp" @@ -13,6 +14,10 @@ XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_comm, s4u_activity, "S4U asynchronous commun namespace simgrid { namespace s4u { +xbt::signal Comm::on_sender_start; +xbt::signal Comm::on_receiver_start; +xbt::signal Comm::on_completion; + Comm::~Comm() { if (state_ == State::STARTED && not detached_ && (pimpl_ == nullptr || pimpl_->state_ == SIMIX_RUNNING)) { @@ -25,79 +30,75 @@ Comm::~Comm() } } -int Comm::wait_any_for(std::vector* comms_in, double timeout) +int Comm::wait_any_for(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::unique_ptr rcomms(new kernel::activity::CommImpl*[comms->size()]); + std::transform(begin(*comms), end(*comms), rcomms.get(), + [](const CommPtr& comm) { return static_cast(comm->pimpl_.get()); }); + return simcall_comm_waitany(rcomms.get(), comms->size(), timeout); } void Comm::wait_all(std::vector* comms) { // TODO: this should be a simcall or something // TODO: we are missing a version with timeout - for (CommPtr comm : *comms) { + for (CommPtr comm : *comms) comm->wait(); - } } -Activity* Comm::set_rate(double rate) +CommPtr Comm::set_rate(double rate) { - xbt_assert(state_ == State::INITED); + xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", + __FUNCTION__); rate_ = rate; return this; } -Activity* Comm::set_src_data(void* buff) +CommPtr Comm::set_src_data(void* buff) { - xbt_assert(state_ == State::INITED); + xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", + __FUNCTION__); xbt_assert(dst_buff_ == nullptr, "Cannot set the src and dst buffers at the same time"); src_buff_ = buff; return this; } -Activity* Comm::set_src_data_size(size_t size) + +CommPtr Comm::set_src_data_size(size_t size) { - xbt_assert(state_ == State::INITED); + xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", + __FUNCTION__); src_buff_size_ = size; return this; } -Activity* Comm::set_src_data(void* buff, size_t size) + +CommPtr Comm::set_src_data(void* buff, size_t size) { - xbt_assert(state_ == State::INITED); + xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", + __FUNCTION__); xbt_assert(dst_buff_ == nullptr, "Cannot set the src and dst buffers at the same time"); src_buff_ = buff; src_buff_size_ = size; return this; } -Activity* Comm::set_dst_data(void** buff) +CommPtr Comm::set_dst_data(void** buff) { - xbt_assert(state_ == State::INITED); + xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", + __FUNCTION__); xbt_assert(src_buff_ == nullptr, "Cannot set the src and dst buffers at the same time"); dst_buff_ = buff; return this; } + size_t Comm::get_dst_data_size() { - xbt_assert(state_ == State::FINISHED); + xbt_assert(state_ == State::FINISHED, "You cannot use %s before your communication terminated", __FUNCTION__); return dst_buff_size_; } -Activity* Comm::set_dst_data(void** buff, size_t size) +CommPtr Comm::set_dst_data(void** buff, size_t size) { - xbt_assert(state_ == State::INITED); + xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", + __FUNCTION__); xbt_assert(src_buff_ == nullptr, "Cannot set the src and dst buffers at the same time"); dst_buff_ = buff; @@ -105,17 +106,27 @@ Activity* Comm::set_dst_data(void** buff, size_t size) return this; } -Activity* Comm::start() +CommPtr Comm::set_tracing_category(const std::string& category) { - xbt_assert(state_ == State::INITED); + xbt_assert(state_ == State::INITED, "Cannot change the tracing category of an exec after its start"); + tracing_category_ = category; + return this; +} + +Comm* Comm::start() +{ + xbt_assert(get_state() == State::INITED, "You cannot use %s() once your communication started (not implemented)", + __FUNCTION__); if (src_buff_ != nullptr) { // Sender side + on_sender_start(*Actor::self()); 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()); 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"); @@ -125,38 +136,45 @@ Activity* Comm::start() } /** @brief Block the calling actor until the communication is finished */ -Activity* Comm::wait() +Comm* Comm::wait() { - return this->wait(-1); + return this->wait_for(-1); } /** @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. */ -Activity* Comm::wait(double timeout) +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 if (src_buff_ != nullptr) { + on_sender_start(*Actor::self()); 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()); 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; + break; case State::STARTED: simcall_comm_wait(pimpl_, timeout); + on_completion(*Actor::self()); state_ = State::FINISHED; - return this; + break; + + case State::CANCELED: + throw CancelException(XBT_THROW_POINT, "Communication canceled"); default: THROW_IMPOSSIBLE; @@ -165,28 +183,28 @@ Activity* Comm::wait(double timeout) } int Comm::test_any(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::unique_ptr rcomms(new kernel::activity::CommImpl*[comms->size()]); + std::transform(begin(*comms), end(*comms), rcomms.get(), + [](const CommPtr& comm) { return static_cast(comm->pimpl_.get()); }); + return simcall_comm_testany(rcomms.get(), comms->size()); } -Activity* Comm::detach() +Comm* Comm::detach() { - xbt_assert(state_ == State::INITED, "You cannot detach communications once they are started (not implemented)."); + xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", + __FUNCTION__); xbt_assert(src_buff_ != nullptr && src_buff_size_ != 0, "You can only detach sends, not recvs"); detached_ = true; return start(); } -Activity* Comm::cancel() +Comm* Comm::cancel() { - simgrid::kernel::activity::CommImplPtr commPimpl = - boost::static_pointer_cast(pimpl_); - commPimpl->cancel(); + kernel::actor::simcall([this] { + if (pimpl_) + boost::static_pointer_cast(pimpl_)->cancel(); + }); + state_ = State::CANCELED; return this; } @@ -207,21 +225,15 @@ bool Comm::test() return false; } -MailboxPtr Comm::get_mailbox() +Mailbox* Comm::get_mailbox() { return mailbox_; } -void intrusive_ptr_release(simgrid::s4u::Comm* c) +Actor* Comm::get_sender() { - if (c->refcount_.fetch_sub(1, std::memory_order_release) == 1) { - std::atomic_thread_fence(std::memory_order_acquire); - delete c; - } -} -void intrusive_ptr_add_ref(simgrid::s4u::Comm* c) -{ - c->refcount_.fetch_add(1, std::memory_order_relaxed); + return sender_ ? sender_->ciface() : nullptr; } + } // namespace s4u } // namespace simgrid