X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/9ab646c308891c340270d62264b0d08354a1302b..b372d23d3d8818b4547ef207f43b45709298a002:/src/s4u/s4u_Comm.cpp diff --git a/src/s4u/s4u_Comm.cpp b/src/s4u/s4u_Comm.cpp index 8ccb8628d5..55a79e5aa7 100644 --- a/src/s4u/s4u_Comm.cpp +++ b/src/s4u/s4u_Comm.cpp @@ -10,13 +10,14 @@ #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() { @@ -33,10 +34,10 @@ Comm::~Comm() 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()); }); - int changed_pos = 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; @@ -95,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_; @@ -111,31 +112,28 @@ CommPtr 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(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_, 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_, 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; } @@ -161,12 +159,12 @@ Comm* Comm::wait_for(double timeout) 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_, 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_, get_user_data(), timeout, rate_); } @@ -175,8 +173,7 @@ Comm* Comm::wait_for(double timeout) 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; @@ -187,15 +184,16 @@ Comm* Comm::wait_for(double timeout) default: THROW_IMPOSSIBLE; } + on_completion(*this); return this; } 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()); }); - int changed_pos = 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; @@ -223,7 +221,8 @@ 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; @@ -231,7 +230,7 @@ bool Comm::test() 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; @@ -239,15 +238,97 @@ bool Comm::test() return false; } -Mailbox* Comm::get_mailbox() +Mailbox* Comm::get_mailbox() const { return mailbox_; } -Actor* Comm::get_sender() +Actor* Comm::get_sender() const { - return sender_ ? sender_->ciface() : nullptr; + 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) +{ + 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; +} + +sg_error_t sg_comm_wait_for(sg_comm_t comm, double timeout) +{ + 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 sg_comm_wait_all(sg_comm_t* comms, size_t count) +{ + 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; +}