X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/d9b336fd19ff354bdf49ff12183e92e678da007a..94f76952308050e13df40d4df3852da353f73f35:/src/s4u/s4u_Comm.cpp diff --git a/src/s4u/s4u_Comm.cpp b/src/s4u/s4u_Comm.cpp index 8d747ed1a6..bcf557de35 100644 --- a/src/s4u/s4u_Comm.cpp +++ b/src/s4u/s4u_Comm.cpp @@ -8,6 +8,7 @@ #include "simgrid/Exception.hpp" #include "simgrid/s4u/Comm.hpp" +#include "simgrid/s4u/Engine.hpp" #include "simgrid/s4u/Mailbox.hpp" #include @@ -38,25 +39,45 @@ Comm::~Comm() } } -int Comm::wait_any_for(const std::vector* comms, double timeout) +ssize_t Comm::wait_any_for(const std::vector& comms, double timeout) { - std::vector rcomms(comms->size()); - std::transform(begin(*comms), end(*comms), begin(rcomms), + 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.data(), rcomms.size(), timeout); + ssize_t changed_pos = simcall_comm_waitany(rcomms.data(), rcomms.size(), timeout); if (changed_pos != -1) - comms->at(changed_pos)->complete(State::FINISHED); + comms.at(changed_pos)->complete(State::FINISHED); return changed_pos; } -void Comm::wait_all(const std::vector* comms) +void Comm::wait_all(const std::vector& comms) { // TODO: this should be a simcall or something - // TODO: we are missing a version with timeout - for (CommPtr comm : *comms) + for (auto& comm : comms) comm->wait(); } +size_t Comm::wait_all_for(const std::vector& comms, double timeout) +{ + if (timeout < 0.0) { + wait_all(comms); + return comms.size(); + } + + double deadline = Engine::get_clock() + timeout; + std::vector waited_comm(1, nullptr); + for (size_t i = 0; i < comms.size(); i++) { + double wait_timeout = std::max(0.0, deadline - Engine::get_clock()); + waited_comm[0] = comms[i]; + // Using wait_any_for() here (and not wait_for) because we don't want comms to be invalidated on timeout + if (wait_any_for(waited_comm, wait_timeout) == -1) { + XBT_DEBUG("Timeout (%g): i = %zu", wait_timeout, i); + return i; + } + } + return comms.size(); +} + CommPtr Comm::set_rate(double rate) { xbt_assert(state_ == State::INITED, "You cannot use %s() once your communication started (not implemented)", @@ -182,12 +203,6 @@ Comm* Comm::start() return this; } -/** @brief Block the calling actor until the communication is finished */ -Comm* Comm::wait() -{ - return this->wait_for(-1); -} - /** @brief Block the calling actor until the communication is finished, or until timeout * * On timeout, an exception is thrown and the communication is invalidated. @@ -230,14 +245,14 @@ Comm* Comm::wait_for(double timeout) return this; } -int Comm::test_any(const std::vector* comms) +ssize_t Comm::test_any(const std::vector& comms) { - std::vector rcomms(comms->size()); - std::transform(begin(*comms), end(*comms), begin(rcomms), + 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.data(), rcomms.size()); + ssize_t changed_pos = simcall_comm_testany(rcomms.data(), rcomms.size()); if (changed_pos != -1) - comms->at(changed_pos)->complete(State::FINISHED); + comms.at(changed_pos)->complete(State::FINISHED); return changed_pos; } @@ -282,6 +297,12 @@ Actor* Comm::get_sender() const return sender ? sender->get_ciface() : nullptr; } +CommPtr Comm::set_copy_data_callback(void (*callback)(kernel::activity::CommImpl*, void*, size_t)) +{ + copy_data_function_ = callback; + return this; +} + } // namespace s4u } // namespace simgrid /* **************************** Public C interface *************************** */ @@ -304,19 +325,7 @@ int sg_comm_test(sg_comm_t comm) 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; + return sg_comm_wait_for(comm, -1); } sg_error_t sg_comm_wait_for(sg_comm_t comm, double timeout) @@ -337,28 +346,36 @@ sg_error_t sg_comm_wait_for(sg_comm_t comm, double timeout) } void sg_comm_wait_all(sg_comm_t* comms, size_t count) +{ + sg_comm_wait_all_for(comms, count, -1); +} + +size_t sg_comm_wait_all_for(sg_comm_t* comms, size_t count, double timeout) { std::vector s4u_comms; - for (unsigned int i = 0; i < count; i++) + for (size_t i = 0; i < count; i++) s4u_comms.emplace_back(comms[i], false); - simgrid::s4u::Comm::wait_all(&s4u_comms); + size_t pos = simgrid::s4u::Comm::wait_all_for(s4u_comms, timeout); + for (size_t i = pos; i < count; i++) + s4u_comms[i]->add_ref(); + return pos; } -int sg_comm_wait_any(sg_comm_t* comms, size_t count) +ssize_t 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) +ssize_t 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++) + for (size_t 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) + ssize_t pos = simgrid::s4u::Comm::wait_any_for(s4u_comms, timeout); + for (size_t i = 0; i < count; i++) { + if (pos != -1 && static_cast(pos) != i) s4u_comms[i]->add_ref(); } return pos;