X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/a47e722190af3bc98a0d5bd16807cb3939f6e4af..1b94b2100679d5b4355fccb6d22927efc12ca5bc:/src/s4u/s4u_Comm.cpp diff --git a/src/s4u/s4u_Comm.cpp b/src/s4u/s4u_Comm.cpp index 14741370c9..a841ac44ac 100644 --- a/src/s4u/s4u_Comm.cpp +++ b/src/s4u/s4u_Comm.cpp @@ -1,11 +1,13 @@ -/* 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. */ #include "src/msg/msg_private.hpp" +#include "src/simix/ActorImpl.hpp" #include "xbt/log.h" +#include "simgrid/Exception.hpp" #include "simgrid/s4u/Comm.hpp" #include "simgrid/s4u/Mailbox.hpp" @@ -29,24 +31,14 @@ 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); + std::unique_ptr rcomms( + new simgrid::kernel::activity::CommImpl*[comms->size()]); + std::transform(begin(*comms), end(*comms), rcomms.get(), [](const CommPtr& comm) { + return static_cast(comm->pimpl_.get()); }); - 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; + return simcall_comm_waitany(rcomms.get(), comms->size(), timeout); } void Comm::wait_all(std::vector* comms) @@ -57,50 +49,59 @@ void Comm::wait_all(std::vector* 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; @@ -108,9 +109,10 @@ Activity* Comm::set_dst_data(void** buff, size_t size) return this; } -Activity* Comm::start() +Comm* Comm::start() { - xbt_assert(state_ == State::INITED); + xbt_assert(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()); @@ -130,7 +132,7 @@ Activity* Comm::start() } /** @brief Block the calling actor until the communication is finished */ -Activity* Comm::wait() +Comm* Comm::wait() { return this->wait_for(-1); } @@ -141,11 +143,11 @@ Activity* Comm::wait() * * @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_for(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) { @@ -159,13 +161,16 @@ Activity* Comm::wait_for(double timeout) 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; @@ -174,26 +179,29 @@ Activity* Comm::wait_for(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 simgrid::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::simix::simcall([this] { dynamic_cast(pimpl_.get())->cancel(); }); + simgrid::simix::simcall([this] { + if (pimpl_) + boost::static_pointer_cast(pimpl_)->cancel(); + }); state_ = State::CANCELED; return this; } @@ -220,6 +228,11 @@ MailboxPtr Comm::get_mailbox() return mailbox_; } +ActorPtr Comm::get_sender() +{ + return sender_ ? sender_->iface() : nullptr; +} + void intrusive_ptr_release(simgrid::s4u::Comm* c) { if (c->refcount_.fetch_sub(1, std::memory_order_release) == 1) {