X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/27a6474a3824ee6d14b321a6f08fadafb7b430ab..9b5c287fbf93c2ae7c3d18c8584647ef9920fe87:/src/s4u/s4u_Comm.cpp diff --git a/src/s4u/s4u_Comm.cpp b/src/s4u/s4u_Comm.cpp index 658097ea50..730a6b4f29 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,9 +14,9 @@ XBT_LOG_NEW_DEFAULT_SUBCATEGORY(s4u_comm, s4u_activity, "S4U asynchronous commun 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_sender_start; +xbt::signal Comm::on_receiver_start; +xbt::signal Comm::on_completion; Comm::~Comm() { @@ -29,24 +30,12 @@ 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) @@ -57,7 +46,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 +54,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 +62,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 +81,7 @@ 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 +89,13 @@ Comm* Comm::set_dst_data(void** buff) dst_buff_ = buff; return this; } + size_t Comm::get_dst_data_size() { 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__); @@ -114,18 +106,25 @@ Comm* Comm::set_dst_data(void** buff, size_t size) return this; } +CommPtr Comm::set_tracing_category(const std::string& category) +{ + 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(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()); + 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_); } else if (dst_buff_ != nullptr) { // Receiver side xbt_assert(not detached_, "Receive cannot be detached"); - on_receiver_start(Actor::self()); + 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_); @@ -144,7 +143,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,27 +151,30 @@ 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()); + 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); } else { // Receiver - on_receiver_start(Actor::self()); + 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_); } state_ = State::FINISHED; - return this; + break; case State::STARTED: simcall_comm_wait(pimpl_, timeout); - on_completion(Actor::self()); + on_completion(*Actor::self()); state_ = State::FINISHED; - return this; + break; + + case State::CANCELED: + throw CancelException(XBT_THROW_POINT, "Communication canceled"); default: THROW_IMPOSSIBLE; @@ -181,13 +183,10 @@ Comm* 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 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()); } Comm* Comm::detach() @@ -201,7 +200,10 @@ Comm* Comm::detach() 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; } @@ -223,11 +225,16 @@ bool Comm::test() return false; } -MailboxPtr Comm::get_mailbox() +Mailbox* Comm::get_mailbox() { return mailbox_; } +Actor* Comm::get_sender() +{ + return sender_ ? sender_->ciface() : nullptr; +} + void intrusive_ptr_release(simgrid::s4u::Comm* c) { if (c->refcount_.fetch_sub(1, std::memory_order_release) == 1) {