X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/a769d0bfd8cf7d4e8f9ae4dfbcf69d8a851508da..b372d23d3d8818b4547ef207f43b45709298a002:/src/s4u/s4u_Comm.cpp diff --git a/src/s4u/s4u_Comm.cpp b/src/s4u/s4u_Comm.cpp index 16b1284b19..55a79e5aa7 100644 --- a/src/s4u/s4u_Comm.cpp +++ b/src/s4u/s4u_Comm.cpp @@ -1,4 +1,4 @@ -/* Copyright (c) 2006-2019. 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. */ @@ -10,35 +10,40 @@ #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 { -xbt::signal Comm::on_sender_start; -xbt::signal Comm::on_receiver_start; -xbt::signal 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, double timeout) +int Comm::wait_any_for(const std::vector* comms, double timeout) { - std::unique_ptr rcomms(new kernel::activity::CommImpl*[comms->size()]); - std::transform(begin(*comms), end(*comms), rcomms.get(), + std::vector rcomms(comms->size()); + std::transform(begin(*comms), end(*comms), begin(rcomms), [](const CommPtr& comm) { return static_cast(comm->pimpl_.get()); }); - return simcall_comm_waitany(rcomms.get(), comms->size(), timeout); + 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 @@ -81,6 +86,7 @@ CommPtr Comm::set_src_data(void* buff, size_t size) src_buff_size_ = size; return this; } + CommPtr Comm::set_dst_data(void** buff) { xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", @@ -90,7 +96,7 @@ CommPtr Comm::set_dst_data(void** 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_; @@ -108,22 +114,26 @@ CommPtr 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; } @@ -136,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. */ @@ -146,24 +156,26 @@ Comm* Comm::wait_for(double timeout) case State::FINISHED: 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; + 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; + this->release_dependencies(); break; case State::CANCELED: @@ -172,14 +184,19 @@ Comm* Comm::wait_for(double timeout) default: THROW_IMPOSSIBLE; } + on_completion(*this); return this; } -int Comm::test_any(std::vector* comms) + +int Comm::test_any(const std::vector* comms) { - std::unique_ptr rcomms(new kernel::activity::CommImpl*[comms->size()]); - std::transform(begin(*comms), end(*comms), rcomms.get(), + std::vector rcomms(comms->size()); + std::transform(begin(*comms), end(*comms), begin(rcomms), [](const CommPtr& comm) { return static_cast(comm->pimpl_.get()); }); - return simcall_comm_testany(rcomms.get(), comms->size()); + 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() @@ -188,12 +205,13 @@ 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() { - simix::simcall([this] { + kernel::actor::simcall([this] { if (pimpl_) boost::static_pointer_cast(pimpl_)->cancel(); }); @@ -203,41 +221,114 @@ Comm* Comm::cancel() 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; } -Mailbox* Comm::get_mailbox() +Mailbox* Comm::get_mailbox() const { return mailbox_; } -ActorPtr Comm::get_sender() +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) { - return sender_ ? sender_->iface() : nullptr; + 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_release(simgrid::s4u::Comm* c) +sg_error_t sg_comm_wait_for(sg_comm_t comm, double timeout) { - 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(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 intrusive_ptr_add_ref(simgrid::s4u::Comm* c) + +void sg_comm_wait_all(sg_comm_t* comms, size_t count) { - c->refcount_.fetch_add(1, std::memory_order_relaxed); + 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