X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/89047acf6125582321f8e86f9765b0f57e350e2c..b8df87e176f27b25534f27d7e240defa32ca35bc:/src/s4u/s4u_Comm.cpp diff --git a/src/s4u/s4u_Comm.cpp b/src/s4u/s4u_Comm.cpp index e9c1692b35..a5e19da341 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. */ @@ -13,9 +13,13 @@ 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; + Comm::~Comm() { - if (state_ == State::started && not detached_ && (pimpl_ == nullptr || pimpl_->state_ == SIMIX_RUNNING)) { + if (state_ == State::STARTED && not detached_ && (pimpl_ == nullptr || pimpl_->state_ == SIMIX_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_); @@ -32,9 +36,9 @@ int Comm::wait_any_for(std::vector* comms_in, double timeout) intrusive_ptr_release(*(simgrid::kernel::activity::ActivityImpl**)ptr); }); for (auto const& comm : *comms_in) { - if (comm->state_ == Activity::State::inited) + if (comm->state_ == Activity::State::INITED) comm->start(); - xbt_assert(comm->state_ == Activity::State::started); + 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); @@ -49,55 +53,60 @@ 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) +Comm* 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) +Comm* 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) +Comm* 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) +Comm* 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) +Comm* 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) +Comm* 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,29 +114,32 @@ 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()); 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()); pimpl_ = simcall_comm_irecv(receiver_, mailbox_->get_impl(), dst_buff_, &dst_buff_size_, match_fun_, copy_data_function_, user_data_, rate_); } else { xbt_die("Cannot start a communication before specifying whether we are the sender or the receiver"); } - state_ = State::started; + state_ = State::STARTED; return this; } /** @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 @@ -136,26 +148,30 @@ 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(double timeout) +Comm* Comm::wait_for(double timeout) { switch (state_) { - case State::finished: + case State::FINISHED: return this; - case State::inited: // It's not started yet. Do it in one simcall + 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); + } 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_); } - state_ = State::finished; + state_ = State::FINISHED; return this; - case State::started: + case State::STARTED: simcall_comm_wait(pimpl_, timeout); - state_ = State::finished; + on_completion(Actor::self()); + state_ = State::FINISHED; return this; default: @@ -174,34 +190,34 @@ int Comm::test_any(std::vector* comms) return res; } -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(); + simgrid::simix::simcall([this] { dynamic_cast(pimpl_.get())->cancel(); }); + state_ = State::CANCELED; return this; } bool Comm::test() { - xbt_assert(state_ == State::inited || state_ == State::started || state_ == State::finished); + xbt_assert(state_ == State::INITED || state_ == State::STARTED || state_ == State::FINISHED); - if (state_ == State::finished) + if (state_ == State::FINISHED) return true; - if (state_ == State::inited) + if (state_ == State::INITED) this->start(); if (simcall_comm_test(pimpl_)) { - state_ = State::finished; + state_ = State::FINISHED; return true; } return false;