X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/2da658602553ee14a2d2e3d7d6f588a512218bed..e6a1650c1da9d6a60f8fc5d12ebce854120ccdb7:/src/s4u/s4u_Comm.cpp diff --git a/src/s4u/s4u_Comm.cpp b/src/s4u/s4u_Comm.cpp index 33c613e69a..6f2602c31a 100644 --- a/src/s4u/s4u_Comm.cpp +++ b/src/s4u/s4u_Comm.cpp @@ -35,7 +35,7 @@ Comm::~Comm() int Comm::wait_any_for(const std::vector* comms, double timeout) { - std::unique_ptr rcomms(new kernel::activity::CommImpl*[comms->size()]); + auto rcomms = std::make_unique(comms->size()); std::transform(begin(*comms), end(*comms), rcomms.get(), [](const CommPtr& comm) { return static_cast(comm->pimpl_.get()); }); int changed_pos = simcall_comm_waitany(rcomms.get(), comms->size(), timeout); @@ -138,6 +138,10 @@ Comm* Comm::start() } 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; } @@ -194,7 +198,7 @@ Comm* Comm::wait_for(double timeout) int Comm::test_any(const std::vector* comms) { - std::unique_ptr rcomms(new kernel::activity::CommImpl*[comms->size()]); + auto rcomms = std::make_unique(comms->size()); std::transform(begin(*comms), end(*comms), rcomms.get(), [](const CommPtr& comm) { return static_cast(comm->pimpl_.get()); }); int changed_pos = simcall_comm_testany(rcomms.get(), comms->size()); @@ -249,21 +253,73 @@ Mailbox* Comm::get_mailbox() const 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]); + s4u_comms.emplace_back(comms[i], false); simgrid::s4u::Comm::wait_all(&s4u_comms); - for (unsigned int i = 0; i < count; i++) - s4u_comms[i]->unref(); } int sg_comm_wait_any(sg_comm_t* comms, size_t count) @@ -274,11 +330,13 @@ int sg_comm_wait_any(sg_comm_t* comms, size_t count) 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]); - } + 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); - if (pos != -1) - s4u_comms[pos]->unref(); + for (unsigned i = 0; i < count; i++) { + if (pos != -1 && static_cast(pos) != i) + s4u_comms[i]->add_ref(); + } return pos; }