X-Git-Url: http://bilbo.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/d6956e89654083209215c97fe1dc549621204224..cdab1fefaf41fe170f2382758e507aa2e34521e0:/src/s4u/s4u_Comm.cpp diff --git a/src/s4u/s4u_Comm.cpp b/src/s4u/s4u_Comm.cpp index bf135b4d04..46515ea934 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,12 +253,24 @@ 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->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(); @@ -263,21 +279,47 @@ int sg_comm_test(sg_comm_t comm) return finished; } -void sg_comm_wait(sg_comm_t comm) +sg_error_t sg_comm_wait(sg_comm_t comm) { - comm->wait_for(-1); - comm->unref(); + 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) @@ -288,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; }