X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/c3789d017953e10867f46f196b546ec7bd16ab75..cec49bceba2427c45f184c27838ae3715344ca3d:/src/kernel/activity/CommImpl.cpp diff --git a/src/kernel/activity/CommImpl.cpp b/src/kernel/activity/CommImpl.cpp index ae219f6e45..8d6c7e37bd 100644 --- a/src/kernel/activity/CommImpl.cpp +++ b/src/kernel/activity/CommImpl.cpp @@ -19,7 +19,7 @@ XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix, "SIMIX network-related synchronization"); XBT_PRIVATE void simcall_HANDLER_comm_send(smx_simcall_t simcall, smx_actor_t src, smx_mailbox_t mbox, double task_size, - double rate, void* src_buff, size_t src_buff_size, + double rate, unsigned char* src_buff, size_t src_buff_size, int (*match_fun)(void*, void*, simgrid::kernel::activity::CommImpl*), void (*copy_data_fun)(simgrid::kernel::activity::CommImpl*, void*, size_t), void* data, double timeout) @@ -31,8 +31,8 @@ XBT_PRIVATE void simcall_HANDLER_comm_send(smx_simcall_t simcall, smx_actor_t sr } XBT_PRIVATE smx_activity_t simcall_HANDLER_comm_isend( - smx_simcall_t /*simcall*/, smx_actor_t src_proc, smx_mailbox_t mbox, double task_size, double rate, void* src_buff, - size_t src_buff_size, int (*match_fun)(void*, void*, simgrid::kernel::activity::CommImpl*), + smx_simcall_t /*simcall*/, smx_actor_t src_proc, smx_mailbox_t mbox, double task_size, double rate, + unsigned char* src_buff, size_t src_buff_size, int (*match_fun)(void*, void*, simgrid::kernel::activity::CommImpl*), void (*clean_fun)(void*), // used to free the synchro in case of problem after a detached send void (*copy_data_fun)(simgrid::kernel::activity::CommImpl*, void*, size_t), // used to copy data if not default one void* data, bool detached) @@ -73,7 +73,7 @@ XBT_PRIVATE smx_activity_t simcall_HANDLER_comm_isend( } if (detached) { - other_comm->detached_ = true; + other_comm->detach(); other_comm->clean_fun = clean_fun; } else { other_comm->clean_fun = nullptr; @@ -97,7 +97,7 @@ XBT_PRIVATE smx_activity_t simcall_HANDLER_comm_isend( } XBT_PRIVATE void simcall_HANDLER_comm_recv(smx_simcall_t simcall, smx_actor_t receiver, smx_mailbox_t mbox, - void* dst_buff, size_t* dst_buff_size, + unsigned char* dst_buff, size_t* dst_buff_size, int (*match_fun)(void*, void*, simgrid::kernel::activity::CommImpl*), void (*copy_data_fun)(simgrid::kernel::activity::CommImpl*, void*, size_t), void* data, double timeout, double rate) @@ -109,7 +109,7 @@ XBT_PRIVATE void simcall_HANDLER_comm_recv(smx_simcall_t simcall, smx_actor_t re } XBT_PRIVATE smx_activity_t simcall_HANDLER_comm_irecv( - smx_simcall_t /*simcall*/, smx_actor_t receiver, smx_mailbox_t mbox, void* dst_buff, size_t* dst_buff_size, + smx_simcall_t /*simcall*/, smx_actor_t receiver, smx_mailbox_t mbox, unsigned char* dst_buff, size_t* dst_buff_size, simix_match_func_t match_fun, void (*copy_data_fun)(simgrid::kernel::activity::CommImpl*, void*, size_t), void* data, double rate) { @@ -137,8 +137,7 @@ XBT_PRIVATE smx_activity_t simcall_HANDLER_comm_irecv( if (other_comm->surf_action_ && other_comm->get_remaining() < 1e-12) { XBT_DEBUG("comm %p has been already sent, and is finished, destroy it", other_comm.get()); other_comm->state_ = SIMIX_DONE; - other_comm->set_type(simgrid::kernel::activity::CommImpl::Type::DONE); - other_comm->mbox = nullptr; + other_comm->set_type(simgrid::kernel::activity::CommImpl::Type::DONE).set_mailbox(nullptr); } } } else { @@ -189,8 +188,7 @@ void simcall_HANDLER_comm_wait(smx_simcall_t simcall, simgrid::kernel::activity: /* Associate this simcall to the wait synchro */ XBT_DEBUG("simcall_HANDLER_comm_wait, %p", comm); - comm->simcalls_.push_back(simcall); - simcall->issuer->waiting_synchro = comm; + comm->register_simcall(simcall); if (MC_is_active() || MC_record_replay_is_active()) { int idx = SIMCALL_GET_MC_VALUE(simcall); @@ -341,8 +339,8 @@ void SIMIX_comm_copy_buffer_callback(simgrid::kernel::activity::CommImpl* comm, { XBT_DEBUG("Copy the data over"); memcpy(comm->dst_buff_, buff, buff_size); - if (comm->detached_) { // if this is a detached send, the source buffer was duplicated by SMPI sender to make the - // original buffer available to the application ASAP + if (comm->detached()) { // if this is a detached send, the source buffer was duplicated by SMPI sender to make the + // original buffer available to the application ASAP xbt_free(buff); comm->src_buff_ = nullptr; } @@ -380,21 +378,32 @@ CommImpl& CommImpl::set_rate(double rate) rate_ = rate; return *this; } +CommImpl& CommImpl::set_mailbox(MailboxImpl* mbox) +{ + mbox_ = mbox; + return *this; +} -CommImpl& CommImpl::set_src_buff(void* buff, size_t size) +CommImpl& CommImpl::set_src_buff(unsigned char* buff, size_t size) { src_buff_ = buff; src_buff_size_ = size; return *this; } -CommImpl& CommImpl::set_dst_buff(void* buff, size_t* size) +CommImpl& CommImpl::set_dst_buff(unsigned char* buff, size_t* size) { dst_buff_ = buff; dst_buff_size_ = size; return *this; } +CommImpl& CommImpl::detach() +{ + detached_ = true; + return *this; +} + CommImpl::~CommImpl() { XBT_DEBUG("Really free communication %p in state %d (detached = %d)", this, static_cast(state_), detached_); @@ -407,8 +416,8 @@ CommImpl::~CommImpl() if (clean_fun) clean_fun(src_buff_); src_buff_ = nullptr; - } else if (mbox) { - mbox->remove(this); + } else if (mbox_) { + mbox_->remove(this); } } @@ -434,7 +443,7 @@ CommImpl* CommImpl::start() XBT_DEBUG("Communication from '%s' to '%s' failed to start because of a link failure", sender->get_cname(), receiver->get_cname()); state_ = SIMIX_LINK_FAILURE; - cleanupSurf(); + post(); } else if (src_actor_->is_suspended() || dst_actor_->is_suspended()) { /* If any of the process is suspended, create the synchro but stop its execution, @@ -460,7 +469,7 @@ void CommImpl::copy_data() { size_t buff_size = src_buff_size_; /* If there is no data to copy then return */ - if (not src_buff_ || not dst_buff_ || copied) + if (not src_buff_ || not dst_buff_ || copied_) return; XBT_DEBUG("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)", this, @@ -484,7 +493,7 @@ void CommImpl::copy_data() /* Set the copied flag so we copy data only once */ /* (this function might be called from both communication ends) */ - copied = true; + copied_ = true; } void CommImpl::suspend() @@ -508,7 +517,7 @@ void CommImpl::cancel() /* if the synchro is a waiting state means that it is still in a mbox so remove from it and delete it */ if (state_ == SIMIX_WAITING) { if (not detached_) { - mbox->remove(this); + mbox_->remove(this); state_ = SIMIX_CANCELED; } } else if (not MC_is_active() /* when running the MC there are no surf actions */ @@ -589,8 +598,8 @@ void CommImpl::finish() } /* If the synchro is still in a rendez-vous point then remove from it */ - if (mbox) - mbox->remove(this); + if (mbox_) + mbox_->remove(this); XBT_DEBUG("CommImpl::finish(): synchro state = %d", static_cast(state_)); @@ -608,12 +617,12 @@ void CommImpl::finish() case SIMIX_SRC_TIMEOUT: simcall->issuer->exception_ = std::make_exception_ptr( - simgrid::TimeoutError(XBT_THROW_POINT, "Communication timeouted because of the sender")); + simgrid::TimeoutException(XBT_THROW_POINT, "Communication timeouted because of the sender")); break; case SIMIX_DST_TIMEOUT: simcall->issuer->exception_ = std::make_exception_ptr( - simgrid::TimeoutError(XBT_THROW_POINT, "Communication timeouted because of the receiver")); + simgrid::TimeoutException(XBT_THROW_POINT, "Communication timeouted because of the receiver")); break; case SIMIX_SRC_HOST_FAILURE: @@ -682,15 +691,8 @@ void CommImpl::finish() // In order to modify the exception we have to rethrow it: try { std::rethrow_exception(simcall->issuer->exception_); - } catch (simgrid::TimeoutError& e) { - e.value = rank; - simcall->issuer->exception_ = std::make_exception_ptr(e); - } catch (simgrid::NetworkFailureException& e) { - e.value = rank; - simcall->issuer->exception_ = std::make_exception_ptr(e); - } catch (simgrid::CancelException& e) { - e.value = rank; - simcall->issuer->exception_ = std::make_exception_ptr(e); + } catch (simgrid::Exception& e) { + e.value = rank; } }