X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/b6ac23e9dc3754bbad244ed95e7aa386f7be5da6..29036bc0419a7bd0b531370ec066948b49af5602:/src/simix/smx_network.cpp diff --git a/src/simix/smx_network.cpp b/src/simix/smx_network.cpp index 7856a90d19..0b47f73f02 100644 --- a/src/simix/smx_network.cpp +++ b/src/simix/smx_network.cpp @@ -3,8 +3,14 @@ /* This program is free software; you can redistribute it and/or modify it * under the terms of the license (GNU LGPL) which comes with this package. */ +#include + #include +#include + +#include + #include "src/surf/surf_interface.hpp" #include "src/simix/smx_private.h" #include "xbt/log.h" @@ -41,16 +47,10 @@ smx_mailbox_t SIMIX_mbox_create(const char *name) xbt_assert(name, "Mailboxes must have a name"); /* two processes may have pushed the same mbox_create simcall at the same time */ smx_mailbox_t mbox = (smx_mailbox_t) xbt_dict_get_or_null(mailboxes, name); - if (!mbox) { - mbox = xbt_new0(s_smx_mailbox_t, 1); - mbox->name = xbt_strdup(name); - mbox->comm_queue = new std::deque(); - mbox->done_comm_queue = nullptr; // Allocated on need only - mbox->permanent_receiver=NULL; - + mbox = new simgrid::simix::Mailbox(name); XBT_DEBUG("Creating a mailbox at %p with name %s", mbox, name); - xbt_dict_set(mailboxes, mbox->name, mbox, NULL); + xbt_dict_set(mailboxes, mbox->name, mbox, nullptr); } return mbox; } @@ -59,11 +59,7 @@ void SIMIX_mbox_free(void *data) { XBT_DEBUG("mbox free %p", data); smx_mailbox_t mbox = (smx_mailbox_t) data; - xbt_free(mbox->name); - delete mbox->comm_queue; - delete mbox->done_comm_queue; - - xbt_free(mbox); + delete mbox; } smx_mailbox_t SIMIX_mbox_get_by_name(const char *name) @@ -78,9 +74,7 @@ smx_mailbox_t SIMIX_mbox_get_by_name(const char *name) */ void SIMIX_mbox_set_receiver(smx_mailbox_t mbox, smx_process_t process) { - mbox->permanent_receiver=process; - if (mbox->done_comm_queue == nullptr) - mbox->done_comm_queue = new std::deque(); + mbox->permanent_receiver = process; } /** @@ -91,8 +85,7 @@ void SIMIX_mbox_set_receiver(smx_mailbox_t mbox, smx_process_t process) static inline void SIMIX_mbox_push(smx_mailbox_t mbox, smx_synchro_t synchro) { simgrid::simix::Comm *comm = static_cast(synchro); - - mbox->comm_queue->push_back(comm); + mbox->comm_queue.push_back(comm); comm->mbox = mbox; } @@ -105,10 +98,10 @@ void SIMIX_mbox_remove(smx_mailbox_t mbox, smx_synchro_t synchro) { simgrid::simix::Comm *comm = static_cast(synchro); - comm->mbox = NULL; - for (auto it = mbox->comm_queue->begin(); it != mbox->comm_queue->end(); it++) + comm->mbox = nullptr; + for (auto it = mbox->comm_queue.begin(); it != mbox->comm_queue.end(); it++) if (*it == comm) { - mbox->comm_queue->erase(it); + mbox->comm_queue. erase(it); return; } xbt_die("Cannot remove this comm that is not part of the mailbox"); @@ -117,12 +110,12 @@ void SIMIX_mbox_remove(smx_mailbox_t mbox, smx_synchro_t synchro) /** * \brief Checks if there is a communication synchro queued in a deque matching our needs * \param type The type of communication we are looking for (comm_send, comm_recv) - * \return The communication synchro if found, NULL otherwise + * \return The communication synchro if found, nullptr otherwise */ static smx_synchro_t _find_matching_comm(std::deque *deque, e_smx_comm_type_t type, int (*match_fun)(void *, void *,smx_synchro_t), void *this_user_data, smx_synchro_t my_synchro, bool remove_matching) { - void* other_user_data = NULL; + void* other_user_data = nullptr; for(auto it = deque->begin(); it != deque->end(); it++){ smx_synchro_t synchro = *it; @@ -143,7 +136,7 @@ static smx_synchro_t _find_matching_comm(std::deque *deque, e_smx #if HAVE_MC comm->mbox_cpy = comm->mbox; #endif - comm->mbox = NULL; + comm->mbox = nullptr; return comm; } XBT_DEBUG("Sorry, communication synchro %p does not match our needs:" @@ -151,7 +144,7 @@ static smx_synchro_t _find_matching_comm(std::deque *deque, e_smx comm, (int)comm->type, (int)type); } XBT_DEBUG("No matching communication synchro found"); - return NULL; + return nullptr; } /******************************************************************************/ @@ -164,7 +157,7 @@ XBT_PRIVATE void simcall_HANDLER_comm_send(smx_simcall_t simcall, smx_process_t void (*copy_data_fun)(smx_synchro_t, void*, size_t), void *data, double timeout){ smx_synchro_t comm = simcall_HANDLER_comm_isend(simcall, src, mbox, task_size, rate, - src_buff, src_buff_size, match_fun, NULL, copy_data_fun, + src_buff, src_buff_size, match_fun, nullptr, copy_data_fun, data, 0); SIMCALL_SET_MC_VALUE(simcall, 0); simcall_HANDLER_comm_wait(simcall, comm, timeout); @@ -187,7 +180,7 @@ XBT_PRIVATE smx_synchro_t simcall_HANDLER_comm_isend(smx_simcall_t simcall, smx_ * * If it is not found then push our communication into the rendez-vous point */ smx_synchro_t other_synchro = - _find_matching_comm(mbox->comm_queue, SIMIX_COMM_RECEIVE, match_fun, data, this_synchro, /*remove_matching*/true); + _find_matching_comm(&mbox->comm_queue, SIMIX_COMM_RECEIVE, match_fun, data, this_synchro, /*remove_matching*/true); simgrid::simix::Comm *other_comm = static_cast(other_synchro); @@ -195,12 +188,12 @@ XBT_PRIVATE smx_synchro_t simcall_HANDLER_comm_isend(smx_simcall_t simcall, smx_ other_synchro = this_synchro; other_comm = static_cast(other_synchro); - if (mbox->permanent_receiver!=NULL){ + if (mbox->permanent_receiver!=nullptr){ //this mailbox is for small messages, which have to be sent right now other_synchro->state = SIMIX_READY; - other_comm->dst_proc=mbox->permanent_receiver; + other_comm->dst_proc=mbox->permanent_receiver.get(); other_comm->ref(); - mbox->done_comm_queue->push_back(other_synchro); + mbox->done_comm_queue.push_back(other_synchro); other_comm->mbox=mbox; XBT_DEBUG("pushing a message into the permanent receive fifo %p, comm %p", mbox, &(other_comm)); @@ -222,7 +215,7 @@ XBT_PRIVATE smx_synchro_t simcall_HANDLER_comm_isend(smx_simcall_t simcall, smx_ other_comm->detached = true; other_comm->clean_fun = clean_fun; } else { - other_comm->clean_fun = NULL; + other_comm->clean_fun = nullptr; } /* Setup the communication synchro */ @@ -239,11 +232,11 @@ XBT_PRIVATE smx_synchro_t simcall_HANDLER_comm_isend(smx_simcall_t simcall, smx_ if (MC_is_active() || MC_record_replay_is_active()) { other_comm->state = SIMIX_RUNNING; - return (detached ? NULL : other_comm); + return (detached ? nullptr : other_comm); } SIMIX_comm_start(other_comm); - return (detached ? NULL : other_comm); + return (detached ? nullptr : other_comm); } XBT_PRIVATE void simcall_HANDLER_comm_recv(smx_simcall_t simcall, smx_process_t receiver, smx_mailbox_t mbox, @@ -271,16 +264,16 @@ smx_synchro_t SIMIX_comm_irecv(smx_process_t dst_proc, smx_mailbox_t mbox, void void (*copy_data_fun)(smx_synchro_t, void*, size_t), // used to copy data if not default one void *data, double rate) { - XBT_DEBUG("recv from %p %p", mbox, mbox->comm_queue); + XBT_DEBUG("recv from %p %p", mbox, &mbox->comm_queue); simgrid::simix::Comm* this_synchro = new simgrid::simix::Comm(SIMIX_COMM_RECEIVE); smx_synchro_t other_synchro; //communication already done, get it inside the fifo of completed comms - if (mbox->permanent_receiver && ! mbox->done_comm_queue->empty()) { + if (mbox->permanent_receiver != nullptr && ! mbox->done_comm_queue.empty()) { XBT_DEBUG("We have a comm that has probably already been received, trying to match it, to skip the communication"); //find a match in the already received fifo - other_synchro = _find_matching_comm(mbox->done_comm_queue, SIMIX_COMM_SEND, match_fun, data, this_synchro,/*remove_matching*/true); + other_synchro = _find_matching_comm(&mbox->done_comm_queue, SIMIX_COMM_SEND, match_fun, data, this_synchro,/*remove_matching*/true); //if not found, assume the receiver came first, register it to the mailbox in the classical way if (!other_synchro) { XBT_DEBUG("We have messages in the permanent receive list, but not the one we are looking for, pushing request into fifo"); @@ -293,7 +286,7 @@ smx_synchro_t SIMIX_comm_irecv(smx_process_t dst_proc, smx_mailbox_t mbox, void XBT_DEBUG("comm %p has been already sent, and is finished, destroy it",other_comm); other_comm->state = SIMIX_DONE; other_comm->type = SIMIX_COMM_DONE; - other_comm->mbox = NULL; + other_comm->mbox = nullptr; } other_comm->unref(); static_cast(this_synchro)->unref(); @@ -305,10 +298,10 @@ smx_synchro_t SIMIX_comm_irecv(smx_process_t dst_proc, smx_mailbox_t mbox, void * ourself so that the other side also gets a chance of choosing if it wants to match with us. * * If it is not found then push our communication into the rendez-vous point */ - other_synchro = _find_matching_comm(mbox->comm_queue, SIMIX_COMM_SEND, match_fun, data, this_synchro,/*remove_matching*/true); + other_synchro = _find_matching_comm(&mbox->comm_queue, SIMIX_COMM_SEND, match_fun, data, this_synchro,/*remove_matching*/true); if (!other_synchro) { - XBT_DEBUG("Receive pushed first %zu", mbox->comm_queue->size()); + XBT_DEBUG("Receive pushed first %zu", mbox->comm_queue.size()); other_synchro = this_synchro; SIMIX_mbox_push(mbox, this_synchro); } else { @@ -353,7 +346,7 @@ smx_synchro_t simcall_HANDLER_comm_iprobe(smx_simcall_t simcall, smx_mailbox_t m smx_synchro_t SIMIX_comm_iprobe(smx_process_t dst_proc, smx_mailbox_t mbox, int type, int src, int tag, int (*match_fun)(void *, void *, smx_synchro_t), void *data) { - XBT_DEBUG("iprobe from %p %p", mbox, mbox->comm_queue); + XBT_DEBUG("iprobe from %p %p", mbox, &mbox->comm_queue); simgrid::simix::Comm* this_comm; int smx_type; if(type == 1){ @@ -363,15 +356,16 @@ smx_synchro_t SIMIX_comm_iprobe(smx_process_t dst_proc, smx_mailbox_t mbox, int this_comm = new simgrid::simix::Comm(SIMIX_COMM_RECEIVE); smx_type = SIMIX_COMM_SEND; } - smx_synchro_t other_synchro=NULL; - if(mbox->permanent_receiver && ! mbox->done_comm_queue->empty()){ + smx_synchro_t other_synchro=nullptr; + if (mbox->permanent_receiver != nullptr && !mbox->done_comm_queue.empty()) { XBT_DEBUG("first check in the permanent recv mailbox, to see if we already got something"); - other_synchro = - _find_matching_comm(mbox->done_comm_queue, (e_smx_comm_type_t) smx_type, match_fun, data, this_comm,/*remove_matching*/false); + other_synchro = _find_matching_comm(&mbox->done_comm_queue, + (e_smx_comm_type_t) smx_type, match_fun, data, this_comm,/*remove_matching*/false); } if (!other_synchro){ XBT_DEBUG("check if we have more luck in the normal mailbox"); - other_synchro = _find_matching_comm(mbox->comm_queue, (e_smx_comm_type_t) smx_type, match_fun, data, this_comm,/*remove_matching*/false); + other_synchro = _find_matching_comm(&mbox->comm_queue, + (e_smx_comm_type_t) smx_type, match_fun, data, this_comm,/*remove_matching*/false); } if(other_synchro) @@ -451,10 +445,11 @@ void simcall_HANDLER_comm_test(smx_simcall_t simcall, smx_synchro_t synchro) } } -void simcall_HANDLER_comm_testany(smx_simcall_t simcall, xbt_dynar_t synchros) +void simcall_HANDLER_comm_testany( + smx_simcall_t simcall, simgrid::simix::Synchro* comms[], size_t count) { - unsigned int cursor; - smx_synchro_t synchro; + // The default result is -1 -- this means, "nothing is ready". + // It can be changed below, but only if something matches. simcall_comm_testany__set__result(simcall, -1); if (MC_is_active() || MC_record_replay_is_active()){ @@ -462,7 +457,7 @@ void simcall_HANDLER_comm_testany(smx_simcall_t simcall, xbt_dynar_t synchros) if(idx == -1){ SIMIX_simcall_answer(simcall); }else{ - synchro = xbt_dynar_get_as(synchros, idx, smx_synchro_t); + simgrid::simix::Synchro* synchro = comms[idx]; simcall_comm_testany__set__result(simcall, idx); synchro->simcalls.push_back(simcall); synchro->state = SIMIX_DONE; @@ -471,9 +466,10 @@ void simcall_HANDLER_comm_testany(smx_simcall_t simcall, xbt_dynar_t synchros) return; } - xbt_dynar_foreach(simcall_comm_testany__get__comms(simcall), cursor,synchro) { + for (std::size_t i = 0; i != count; ++i) { + simgrid::simix::Synchro* synchro = comms[i]; if (synchro->state != SIMIX_WAITING && synchro->state != SIMIX_RUNNING) { - simcall_comm_testany__set__result(simcall, cursor); + simcall_comm_testany__set__result(simcall, i); synchro->simcalls.push_back(simcall); SIMIX_comm_finish(synchro); return; @@ -603,74 +599,91 @@ void SIMIX_comm_finish(smx_synchro_t synchro) if (simcall->issuer->host->isOff()) { simcall->issuer->context->iwannadie = 1; SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed"); - } else - - switch (synchro->state) { - - case SIMIX_DONE: - XBT_DEBUG("Communication %p complete!", synchro); - SIMIX_comm_copy_data(synchro); - break; - - case SIMIX_SRC_TIMEOUT: - SMX_EXCEPTION(simcall->issuer, timeout_error, 0, "Communication timeouted because of sender"); - break; - - case SIMIX_DST_TIMEOUT: - SMX_EXCEPTION(simcall->issuer, timeout_error, 0, "Communication timeouted because of receiver"); - break; - - case SIMIX_SRC_HOST_FAILURE: - if (simcall->issuer == comm->src_proc) - simcall->issuer->context->iwannadie = 1; -// SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed"); - else - SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed"); - break; - - case SIMIX_DST_HOST_FAILURE: - if (simcall->issuer == comm->dst_proc) - simcall->issuer->context->iwannadie = 1; -// SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed"); - else - SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed"); - break; - - case SIMIX_LINK_FAILURE: - - XBT_DEBUG("Link failure in synchro %p between '%s' and '%s': posting an exception to the issuer: %s (%p) detached:%d", - synchro, - comm->src_proc ? sg_host_get_name(comm->src_proc->host) : NULL, - comm->dst_proc ? sg_host_get_name(comm->dst_proc->host) : NULL, - simcall->issuer->name.c_str(), simcall->issuer, comm->detached); - if (comm->src_proc == simcall->issuer) { - XBT_DEBUG("I'm source"); - } else if (comm->dst_proc == simcall->issuer) { - XBT_DEBUG("I'm dest"); - } else { - XBT_DEBUG("I'm neither source nor dest"); + } else { + switch (synchro->state) { + + case SIMIX_DONE: + XBT_DEBUG("Communication %p complete!", synchro); + SIMIX_comm_copy_data(synchro); + break; + + case SIMIX_SRC_TIMEOUT: + SMX_EXCEPTION(simcall->issuer, timeout_error, 0, "Communication timeouted because of sender"); + break; + + case SIMIX_DST_TIMEOUT: + SMX_EXCEPTION(simcall->issuer, timeout_error, 0, "Communication timeouted because of receiver"); + break; + + case SIMIX_SRC_HOST_FAILURE: + if (simcall->issuer == comm->src_proc) + simcall->issuer->context->iwannadie = 1; + // SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed"); + else + SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed"); + break; + + case SIMIX_DST_HOST_FAILURE: + if (simcall->issuer == comm->dst_proc) + simcall->issuer->context->iwannadie = 1; + // SMX_EXCEPTION(simcall->issuer, host_error, 0, "Host failed"); + else + SMX_EXCEPTION(simcall->issuer, network_error, 0, "Remote peer failed"); + break; + + case SIMIX_LINK_FAILURE: + + XBT_DEBUG("Link failure in synchro %p between '%s' and '%s': posting an exception to the issuer: %s (%p) detached:%d", + synchro, + comm->src_proc ? sg_host_get_name(comm->src_proc->host) : nullptr, + comm->dst_proc ? sg_host_get_name(comm->dst_proc->host) : nullptr, + simcall->issuer->name.c_str(), simcall->issuer, comm->detached); + if (comm->src_proc == simcall->issuer) { + XBT_DEBUG("I'm source"); + } else if (comm->dst_proc == simcall->issuer) { + XBT_DEBUG("I'm dest"); + } else { + XBT_DEBUG("I'm neither source nor dest"); + } + SMX_EXCEPTION(simcall->issuer, network_error, 0, "Link failure"); + break; + + case SIMIX_CANCELED: + if (simcall->issuer == comm->dst_proc) + SMX_EXCEPTION(simcall->issuer, cancel_error, 0, "Communication canceled by the sender"); + else + SMX_EXCEPTION(simcall->issuer, cancel_error, 0, "Communication canceled by the receiver"); + break; + + default: + xbt_die("Unexpected synchro state in SIMIX_comm_finish: %d", (int)synchro->state); } - SMX_EXCEPTION(simcall->issuer, network_error, 0, "Link failure"); - break; - - case SIMIX_CANCELED: - if (simcall->issuer == comm->dst_proc) - SMX_EXCEPTION(simcall->issuer, cancel_error, 0, "Communication canceled by the sender"); - else - SMX_EXCEPTION(simcall->issuer, cancel_error, 0, "Communication canceled by the receiver"); - break; - - default: - xbt_die("Unexpected synchro state in SIMIX_comm_finish: %d", (int)synchro->state); } /* if there is an exception during a waitany or a testany, indicate the position of the failed communication */ - if (simcall->issuer->doexception) { - if (simcall->call == SIMCALL_COMM_WAITANY) { - simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &synchro); + if (simcall->issuer->exception) { + // In order to modify the exception we have to rethrow it: + try { + std::rethrow_exception(simcall->issuer->exception); + } + catch(xbt_ex& e) { + if (simcall->call == SIMCALL_COMM_WAITANY) { + e.value = xbt_dynar_search(simcall_comm_waitany__get__comms(simcall), &synchro); + } + else if (simcall->call == SIMCALL_COMM_TESTANY) { + e.value = -1; + auto comms = simcall_comm_testany__get__comms(simcall); + auto count = simcall_comm_testany__get__count(simcall); + auto element = std::find(comms, comms + count, synchro); + if (element == comms + count) + e.value = -1; + else + e.value = element - comms; + } + simcall->issuer->exception = std::make_exception_ptr(e); } - else if (simcall->call == SIMCALL_COMM_TESTANY) { - simcall->issuer->running_ctx->exception.value = xbt_dynar_search(simcall_comm_testany__get__comms(simcall), &synchro); + catch(...) { + // Nothing to do } } @@ -678,7 +691,7 @@ void SIMIX_comm_finish(smx_synchro_t synchro) simcall->issuer->context->iwannadie = 1; } - simcall->issuer->waiting_synchro = NULL; + simcall->issuer->waiting_synchro = nullptr; xbt_fifo_remove(simcall->issuer->comms, synchro); if(comm->detached){ if(simcall->issuer == comm->src_proc){ @@ -726,7 +739,7 @@ void SIMIX_comm_copy_buffer_callback(smx_synchro_t synchro, void* buff, size_t b 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 xbt_free(buff); - comm->src_buff = NULL; + comm->src_buff = nullptr; } }