X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/27a6474a3824ee6d14b321a6f08fadafb7b430ab..fbf5fafb60eb44c0c00a7283fd05a6dec5f2b58f:/src/s4u/s4u_Comm.cpp diff --git a/src/s4u/s4u_Comm.cpp b/src/s4u/s4u_Comm.cpp index 658097ea50..16b1284b19 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__); @@ -152,7 +144,7 @@ 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) { @@ -166,13 +158,16 @@ Comm* 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; @@ -181,13 +176,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 +193,10 @@ Comm* Comm::detach() Comm* Comm::cancel() { - simgrid::simix::simcall([this] { dynamic_cast(pimpl_.get())->cancel(); }); + simix::simcall([this] { + if (pimpl_) + boost::static_pointer_cast(pimpl_)->cancel(); + }); state_ = State::CANCELED; return this; } @@ -223,11 +218,16 @@ bool Comm::test() return false; } -MailboxPtr Comm::get_mailbox() +Mailbox* 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) {