X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/c7f3c30ef936dacf568b44009d3948c8dc64228c..1372bba9282aabce62d6c3f8bb47c4281337482d:/src/simix/smx_network.c diff --git a/src/simix/smx_network.c b/src/simix/smx_network.c index f1533a8019..29e1bcf118 100644 --- a/src/simix/smx_network.c +++ b/src/simix/smx_network.c @@ -11,13 +11,16 @@ /* Pimple to get an histogram of message sizes in the simulation */ xbt_dict_t msg_sizes = NULL; +#ifdef HAVE_LATENCY_BOUND_TRACKING +xbt_dict_t latency_limited_dict = NULL; +#endif XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix, "Logging specific to SIMIX (network)"); /******************************************************************************/ /* Rendez-Vous Points */ -/******************************************************************************/ +/******************************************************************************/ /** * \brief Creates a new rendez-vous point @@ -41,7 +44,7 @@ smx_rdv_t SIMIX_rdv_create(const char *name) */ void SIMIX_rdv_destroy(smx_rdv_t rdv) { - if(rdv->name) + if (rdv->name) xbt_free(rdv->name); SIMIX_mutex_destroy(rdv->read); SIMIX_mutex_destroy(rdv->write); @@ -54,7 +57,7 @@ void SIMIX_rdv_destroy(smx_rdv_t rdv) * \param rdv The rendez-vous point * \param comm The communication request */ -static inline void SIMIX_rdv_push(smx_rdv_t rdv, smx_comm_t comm) +static XBT_INLINE void SIMIX_rdv_push(smx_rdv_t rdv, smx_comm_t comm) { xbt_fifo_push(rdv->comm_fifo, comm); comm->rdv = rdv; @@ -65,22 +68,23 @@ static inline void SIMIX_rdv_push(smx_rdv_t rdv, smx_comm_t comm) * \param rdv The rendez-vous point * \param comm The communication request */ -static inline void SIMIX_rdv_remove(smx_rdv_t rdv, smx_comm_t comm) +static XBT_INLINE void SIMIX_rdv_remove(smx_rdv_t rdv, smx_comm_t comm) { xbt_fifo_remove(rdv->comm_fifo, comm); comm->rdv = NULL; } - + /** * \brief Checks if there is a communication request queued in a rendez-vous matching our needs * \param type The type of communication we are looking for (comm_send, comm_recv) * \return The communication request if found, NULL otherwise. */ -smx_comm_t SIMIX_rdv_get_request(smx_rdv_t rdv, smx_comm_type_t type) { - smx_comm_t comm = (smx_comm_t)xbt_fifo_get_item_content( - xbt_fifo_get_first_item(rdv->comm_fifo)); +smx_comm_t SIMIX_rdv_get_request(smx_rdv_t rdv, smx_comm_type_t type) +{ + smx_comm_t comm = (smx_comm_t) + xbt_fifo_get_item_content(xbt_fifo_get_first_item(rdv->comm_fifo)); - if(comm && comm->type == type){ + if (comm && comm->type == type) { DEBUG0("Communication request found!"); xbt_fifo_shift(rdv->comm_fifo); SIMIX_communication_use(comm); @@ -99,8 +103,7 @@ smx_comm_t SIMIX_rdv_get_request(smx_rdv_t rdv, smx_comm_type_t type) { * \param host The host to be counted * \return The number of comm request pending in the rdv */ -int -SIMIX_rdv_get_count_waiting_comm(smx_rdv_t rdv, smx_host_t host) +int SIMIX_rdv_get_count_waiting_comm(smx_rdv_t rdv, smx_host_t host) { smx_comm_t comm = NULL; xbt_fifo_item_t item = NULL; @@ -121,27 +124,31 @@ SIMIX_rdv_get_count_waiting_comm(smx_rdv_t rdv, smx_host_t host) */ XBT_INLINE smx_comm_t SIMIX_rdv_get_head(smx_rdv_t rdv) { - return (smx_comm_t)xbt_fifo_get_item_content(xbt_fifo_get_first_item(rdv->comm_fifo)); + return (smx_comm_t) + xbt_fifo_get_item_content(xbt_fifo_get_first_item(rdv->comm_fifo)); } /** @brief adds some API-related data to the rendez-vous point */ -XBT_INLINE void SIMIX_rdv_set_data(smx_rdv_t rdv,void *data) { - rdv->data=data; +XBT_INLINE void SIMIX_rdv_set_data(smx_rdv_t rdv, void *data) +{ + rdv->data = data; } + /** @brief gets API-related data from the rendez-vous point */ -XBT_INLINE void *SIMIX_rdv_get_data(smx_rdv_t rdv) { +XBT_INLINE void *SIMIX_rdv_get_data(smx_rdv_t rdv) +{ return rdv->data; } /******************************************************************************/ /* Communication Requests */ -/******************************************************************************/ +/******************************************************************************/ /** * \brief Creates a new communication request * \param type The type of communication (comm_send, comm_recv) * \return The new communication request - */ + */ smx_comm_t SIMIX_communication_new(smx_comm_type_t type) { /* alloc structures */ @@ -149,7 +156,7 @@ smx_comm_t SIMIX_communication_new(smx_comm_type_t type) comm->type = type; comm->sem = SIMIX_sem_init(0); comm->refcount = 1; - + return comm; } @@ -159,29 +166,47 @@ smx_comm_t SIMIX_communication_new(smx_comm_type_t type) */ void SIMIX_communication_destroy(smx_comm_t comm) { + VERB2("Destroy communication %p; refcount initially %d", comm, + comm->refcount); + +#ifdef HAVE_LATENCY_BOUND_TRACKING + //save is latency limited flag to use afterwards + if (latency_limited_dict == NULL) { + latency_limited_dict = xbt_dict_new(); + } + if (comm->act) { + DEBUG2("adding key %p with latency limited value %d to the dict", comm, + SIMIX_action_is_latency_bounded(comm->act)); + xbt_dicti_set(latency_limited_dict, (uintptr_t) comm, + SIMIX_action_is_latency_bounded(comm->act)); + } +#endif + comm->refcount--; - if(comm->refcount > 0) + if (comm->refcount > 0) return; - if(comm->sem){ + if (comm->sem) { SIMIX_sem_destroy(comm->sem); comm->sem = NULL; } - - if(comm->act){ + + if (comm->act) { SIMIX_action_destroy(comm->act); comm->act = NULL; } - if(comm->src_timeout){ + if (comm->src_timeout) { SIMIX_action_destroy(comm->src_timeout); comm->src_timeout = NULL; } - if(comm->dst_timeout){ - SIMIX_action_destroy(comm->dst_timeout); - comm->dst_timeout = NULL; - } + if (comm->dst_timeout) { + SIMIX_action_destroy(comm->dst_timeout); + comm->dst_timeout = NULL; + } + + xbt_free(comm); } @@ -194,7 +219,7 @@ void SIMIX_communication_destroy(smx_comm_t comm) * maybe is in use by others. * \ */ -static inline void SIMIX_communication_use(smx_comm_t comm) +static XBT_INLINE void SIMIX_communication_use(smx_comm_t comm) { comm->refcount++; } @@ -203,25 +228,26 @@ static inline void SIMIX_communication_use(smx_comm_t comm) * \brief Start the simulation of a communication request * \param comm The communication request */ -static inline void SIMIX_communication_start(smx_comm_t comm) +static XBT_INLINE void SIMIX_communication_start(smx_comm_t comm) { /* If both the sender and the receiver are already there, start the communication */ - if(comm->src_proc && comm->dst_proc){ + if (comm->src_proc && comm->dst_proc) { DEBUG1("Starting communication %p", comm); - comm->act = SIMIX_action_communicate(comm->src_proc->smx_host, - comm->dst_proc->smx_host, NULL, + comm->act = SIMIX_action_communicate(comm->src_proc->smx_host, + comm->dst_proc->smx_host, NULL, comm->task_size, comm->rate); #ifdef HAVE_TRACING - TRACE_smx_action_communicate (comm->act, comm->src_proc); + TRACE_smx_action_communicate(comm->act, comm->src_proc); + TRACE_surf_action(comm->act->surf_action, comm->act->category); #endif /* If any of the process is suspend, create the action but stop its execution, it will be restarted when the sender process resume */ - if(SIMIX_process_is_suspended(comm->src_proc) || - SIMIX_process_is_suspended(comm->dst_proc)) { + if (SIMIX_process_is_suspended(comm->src_proc) || + SIMIX_process_is_suspended(comm->dst_proc)) { SIMIX_action_suspend(comm->act); } - + /* Add the communication as user data of the action */ comm->act->data = comm; @@ -233,7 +259,46 @@ static inline void SIMIX_communication_start(smx_comm_t comm) } /** - * \brief Waits for communication completion and performs error checking + * \brief Performs error checking and cleanup + * \param comm The communication + */ +static XBT_INLINE void SIMIX_communication_cleanup(smx_comm_t comm) +{ + DEBUG1("Checking errors and cleaning communication %p", comm); + + /* Make sure that everyone sleeping on that semaphore is awake, and that nobody will ever block on it */ + SIMIX_sem_release_forever(comm->sem); + + /* Check for errors other than timeouts */ + if (!SIMIX_host_get_state(SIMIX_host_self())) { + if (comm->rdv) + SIMIX_rdv_remove(comm->rdv, comm); + SIMIX_communication_destroy(comm); + THROW0(host_error, 0, "Host failed"); + } else if (SIMIX_action_get_state(comm->act) == SURF_ACTION_FAILED) { + SIMIX_communication_destroy(comm); + THROW0(network_error, 0, "Link failure"); + } else if (!SIMIX_host_get_state(SIMIX_process_get_host(comm->dst_proc)) + || + !SIMIX_host_get_state(SIMIX_process_get_host(comm->src_proc))) + { + /* We test both src&dst because we dunno who we are today, and we already tested myself above. + * So, at the end, we test the remote peer only + * Moreover, we have to test it because if the remote peer fails, the action comm->act is not done nor failed. + * In that case, we got awaken by the little endless actions created in the SIMIX_sem_acquire(comm->sem) + * at the beginning of this function. */ + SIMIX_communication_destroy(comm); + THROW0(network_error, 0, "Remote peer failed"); + + } + /* Copy network data */ + SIMIX_network_copy_data(comm); + + SIMIX_communication_destroy(comm); +} + +/** + * \brief Waits for communication completion * \param comm The communication * \param timeout The max amount of time to wait for the communication to finish * @@ -242,21 +307,27 @@ static inline void SIMIX_communication_start(smx_comm_t comm) * - timeout_error if communication reached the timeout specified (either because of local peer or remote peer) * - network_error if network failed or remote peer failed */ -static inline void SIMIX_communication_wait_for_completion(smx_comm_t comm, double timeout) +static XBT_INLINE void SIMIX_communication_wait_for_completion(smx_comm_t + comm, + double + timeout) { smx_action_t act_sleep = NULL; int src_timeout = 0; int dst_timeout = 0; DEBUG1("Waiting for the completion of communication %p", comm); - + if (timeout >= 0) { act_sleep = SIMIX_action_sleep(SIMIX_host_self(), timeout); - if(SIMIX_process_self()==comm->src_proc) - comm->src_timeout = act_sleep; - else - comm->dst_timeout = act_sleep; - SIMIX_action_set_name(act_sleep,bprintf("Timeout for comm %p and wait on semaphore %p (max_duration:%f)", comm, comm->sem,timeout)); + if (SIMIX_process_self() == comm->src_proc) + comm->src_timeout = act_sleep; + else + comm->dst_timeout = act_sleep; + SIMIX_action_set_name(act_sleep, + bprintf + ("Timeout for comm %p and wait on semaphore %p (max_duration:%f)", + comm, comm->sem, timeout)); SIMIX_register_action_to_semaphore(act_sleep, comm->sem); SIMIX_process_self()->waiting_action = act_sleep; SIMIX_sem_block_onto(comm->sem); @@ -267,13 +338,20 @@ static inline void SIMIX_communication_wait_for_completion(smx_comm_t comm, doub } /* Check for timeouts */ - if ((src_timeout = ((comm->src_timeout) && (SIMIX_action_get_state(comm->src_timeout) == SURF_ACTION_DONE))) || - (dst_timeout = ((comm->dst_timeout) && (SIMIX_action_get_state(comm->dst_timeout) == SURF_ACTION_DONE))) ) { - /* Somebody did a timeout! */ - if (src_timeout) DEBUG1("Communication timeout from the src! %p", comm); - if (dst_timeout) DEBUG1("Communication timeout from the dst! %p", comm); - - if(comm->act && SIMIX_action_get_state(comm->act) == SURF_ACTION_RUNNING) + if ((src_timeout = ((comm->src_timeout) + && (SIMIX_action_get_state(comm->src_timeout) == + SURF_ACTION_DONE))) + || (dst_timeout = ((comm->dst_timeout) + && (SIMIX_action_get_state(comm->dst_timeout) == + SURF_ACTION_DONE)))) { + /* Somebody did a timeout! */ + if (src_timeout) + DEBUG1("Communication timeout from the src! %p", comm); + if (dst_timeout) + DEBUG1("Communication timeout from the dst! %p", comm); + + if (comm->act + && SIMIX_action_get_state(comm->act) == SURF_ACTION_RUNNING) SIMIX_communication_cancel(comm); else if (comm->rdv) SIMIX_rdv_remove(comm->rdv, comm); @@ -282,38 +360,12 @@ static inline void SIMIX_communication_wait_for_completion(smx_comm_t comm, doub SIMIX_sem_release_forever(comm->sem); SIMIX_communication_destroy(comm); - THROW1(timeout_error, 0, "Communication timeouted because of %s",src_timeout?"the source":"the destination"); + THROW1(timeout_error, 0, "Communication timeouted because of %s", + src_timeout ? "the source" : "the destination"); } - DEBUG1("Communication %p complete! Let's check for errors", comm); - - /* Make sure that everyone sleeping on that semaphore is awake, and that nobody will ever block on it */ - SIMIX_sem_release_forever(comm->sem); - - /* Check for errors other than timeouts (they are catched above) */ - if (!SIMIX_host_get_state(SIMIX_host_self())){ - if(comm->rdv) - SIMIX_rdv_remove(comm->rdv, comm); - SIMIX_communication_destroy(comm); - THROW0(host_error, 0, "Host failed"); - } else if (SIMIX_action_get_state(comm->act) == SURF_ACTION_FAILED){ - SIMIX_communication_destroy(comm); - THROW0(network_error, 0, "Link failure"); - } else if (!SIMIX_host_get_state(SIMIX_process_get_host(comm->dst_proc)) || - !SIMIX_host_get_state(SIMIX_process_get_host(comm->src_proc))) { - /* We test both src&dst because we dunno who we are today, and we already tested myself above. - * So, at the end, we test the remote peer only - * Moreover, we have to test it because if the remote peer fails, the action comm->act is not done nor failed. - * In that case, we got awaken by the little endless actions created in the SIMIX_sem_acquire(comm->sem) - * at the beginning of this function. */ - SIMIX_communication_destroy(comm); - THROW0(network_error, 0, "Remote peer failed"); - - } - /* Copy network data */ - SIMIX_network_copy_data(comm); - - SIMIX_communication_destroy(comm); + DEBUG1("Communication %p complete!", comm); + SIMIX_communication_cleanup(comm); } /** @@ -332,24 +384,56 @@ XBT_INLINE void SIMIX_communication_cancel(smx_comm_t comm) */ XBT_INLINE double SIMIX_communication_get_remains(smx_comm_t comm) { + DEBUG1("calling SIMIX_action_get_remains(%p)", comm->act); return SIMIX_action_get_remains(comm->act); -} +} + +#ifdef HAVE_LATENCY_BOUND_TRACKING +/** + * \brief verify if communication is latency bounded + * \param comm The communication + */ +XBT_INLINE int SIMIX_communication_is_latency_bounded(smx_comm_t comm) +{ + //try to find comm on the list of finished flows + uintptr_t key = 0; + uintptr_t data = 0; + xbt_dict_cursor_t cursor; + xbt_dict_foreach(latency_limited_dict, cursor, key, data) { + DEBUG2("comparing key=%p with comm=%p", (void *) key, (void *) comm); + if ((void *) comm == (void *) key) { + DEBUG2("key %p found, return value latency limited value %d", + (void *) key, (int) data); + return (int) data; + } + } + + DEBUG1("calling SIMIX_action_is_latency_bounded(%p)", comm->act); + return SIMIX_action_is_latency_bounded(comm->act); +} +#endif /******************************************************************************/ /* SIMIX_network_copy_data callbacks */ /******************************************************************************/ -static void (*SIMIX_network_copy_data_callback)(smx_comm_t, size_t) = &SIMIX_network_copy_pointer_callback; +static void (*SIMIX_network_copy_data_callback) (smx_comm_t, size_t) = + &SIMIX_network_copy_pointer_callback; -void SIMIX_network_set_copy_data_callback(void (*callback)(smx_comm_t, size_t)) { +void +SIMIX_network_set_copy_data_callback(void (*callback) (smx_comm_t, size_t)) +{ SIMIX_network_copy_data_callback = callback; } -void SIMIX_network_copy_pointer_callback(smx_comm_t comm, size_t buff_size) { - xbt_assert1((buff_size == sizeof(void*)), "Cannot copy %zu bytes: must be sizeof(void*)",buff_size); - *(void**)(comm->dst_buff) = comm->src_buff; +void SIMIX_network_copy_pointer_callback(smx_comm_t comm, size_t buff_size) +{ + xbt_assert1((buff_size == sizeof(void *)), + "Cannot copy %zu bytes: must be sizeof(void*)", buff_size); + *(void **) (comm->dst_buff) = comm->src_buff; } -void SIMIX_network_copy_buffer_callback(smx_comm_t comm, size_t buff_size) { +void SIMIX_network_copy_buffer_callback(smx_comm_t comm, size_t buff_size) +{ memcpy(comm->dst_buff, comm->src_buff, buff_size); } @@ -360,53 +444,58 @@ void SIMIX_network_copy_buffer_callback(smx_comm_t comm, size_t buff_size) { void SIMIX_network_copy_data(smx_comm_t comm) { size_t buff_size = comm->src_buff_size; - + uintptr_t casted_size = 0; + uintptr_t amount = 0; /* If there is no data to be copy then return */ - if(!comm->src_buff || !comm->dst_buff || comm->copied == 1) + if (!comm->src_buff || !comm->dst_buff || comm->copied == 1) return; DEBUG6("Copying comm %p data from %s (%p) -> %s (%p) (%zu bytes)", - comm, - comm->src_proc->smx_host->name, comm->src_buff, - comm->dst_proc->smx_host->name, comm->dst_buff, - buff_size); + comm, + comm->src_proc->smx_host->name, comm->src_buff, + comm->dst_proc->smx_host->name, comm->dst_buff, buff_size); /* Copy at most dst_buff_size bytes of the message to receiver's buffer */ if (comm->dst_buff_size) - buff_size = MIN(buff_size,*(comm->dst_buff_size)); - + buff_size = MIN(buff_size, *(comm->dst_buff_size)); + /* Update the receiver's buffer size to the copied amount */ if (comm->dst_buff_size) *comm->dst_buff_size = buff_size; - if(buff_size == 0) + if (buff_size == 0) return; - (*SIMIX_network_copy_data_callback)(comm, buff_size); + (*SIMIX_network_copy_data_callback) (comm, buff_size); /* Set the copied flag so we copy data only once */ - /* (this function might be called from both communication ends)*/ + /* (this function might be called from both communication ends) */ comm->copied = 1; - + /* pimple to display the message sizes */ { if (msg_sizes == NULL) msg_sizes = xbt_dict_new(); - uintptr_t casted_size = comm->task_size; - uintptr_t amount = xbt_dicti_get(msg_sizes, casted_size); + casted_size = comm->task_size; + amount = xbt_dicti_get(msg_sizes, casted_size); amount++; - xbt_dicti_set(msg_sizes,casted_size, amount); + xbt_dicti_set(msg_sizes, casted_size, amount); } } + #include "xbt.h" /* pimple to display the message sizes */ -void SIMIX_message_sizes_output(const char *filename) { - FILE * out = fopen(filename,"w"); - xbt_assert1(out,"Cannot open file %s",filename); - uintptr_t key,data; +void SIMIX_message_sizes_output(const char *filename) +{ + uintptr_t key = 0; + uintptr_t data = 0; xbt_dict_cursor_t cursor; - xbt_dict_foreach(msg_sizes,cursor,key,data) { - fprintf(out,"%zu %zu\n",key,data); + FILE *out = NULL; + out = fopen(filename, "w"); + xbt_assert1(out, "Cannot open file %s", filename); + + xbt_dict_foreach(msg_sizes, cursor, key, data) { + fprintf(out, "%zu %zu\n", key, data); } fclose(out); } @@ -421,6 +510,26 @@ XBT_INLINE void *SIMIX_communication_get_data(smx_comm_t comm) return comm->data; } +XBT_PUBLIC(void *) SIMIX_communication_get_src_buf(smx_comm_t comm) +{ + return comm->src_buff; +} + +XBT_PUBLIC(void *) SIMIX_communication_get_dst_buf(smx_comm_t comm) +{ + return comm->dst_buff; +} + +XBT_PUBLIC(size_t) SIMIX_communication_get_src_buf_size(smx_comm_t comm) +{ + return comm->src_buff_size; +} + +XBT_PUBLIC(size_t) SIMIX_communication_get_dst_buf_size(smx_comm_t comm) +{ + return *(comm->dst_buff_size); +} + /******************************************************************************/ /* Synchronous Communication */ /******************************************************************************/ @@ -440,12 +549,15 @@ XBT_INLINE void *SIMIX_communication_get_data(smx_comm_t comm) * - timeout_error if communication reached the timeout specified * - network_error if network failed or peer issued a timeout */ -XBT_INLINE void SIMIX_network_send(smx_rdv_t rdv, double task_size, double rate, - double timeout, void *src_buff, size_t src_buff_size, - smx_comm_t *comm_ref, void *data) +XBT_INLINE void SIMIX_network_send(smx_rdv_t rdv, double task_size, + double rate, double timeout, + void *src_buff, size_t src_buff_size, + smx_comm_t * comm_ref, void *data) { - *comm_ref = SIMIX_network_isend(rdv,task_size,rate,src_buff,src_buff_size,data); - SIMIX_network_wait(*comm_ref,timeout); + *comm_ref = + SIMIX_network_isend(rdv, task_size, rate, src_buff, src_buff_size, + data); + SIMIX_network_wait(*comm_ref, timeout); } /** @@ -461,31 +573,34 @@ XBT_INLINE void SIMIX_network_send(smx_rdv_t rdv, double task_size, double rate, * - timeout_error if communication reached the timeout specified * - network_error if network failed or peer issued a timeout */ -XBT_INLINE void SIMIX_network_recv(smx_rdv_t rdv, double timeout, void *dst_buff, - size_t *dst_buff_size, smx_comm_t *comm_ref) +XBT_INLINE void SIMIX_network_recv(smx_rdv_t rdv, double timeout, + void *dst_buff, size_t * dst_buff_size, + smx_comm_t * comm_ref) { - *comm_ref = SIMIX_network_irecv(rdv,dst_buff,dst_buff_size); - SIMIX_network_wait(*comm_ref,timeout); + *comm_ref = + (smx_comm_t) SIMIX_network_irecv(rdv, dst_buff, dst_buff_size); + SIMIX_network_wait(*comm_ref, timeout); } /******************************************************************************/ /* Asynchronous Communication */ /******************************************************************************/ -smx_comm_t SIMIX_network_isend(smx_rdv_t rdv, double task_size, double rate, - void *src_buff, size_t src_buff_size, void *data) +smx_comm_t SIMIX_network_isend(smx_rdv_t rdv, double task_size, + double rate, void *src_buff, + size_t src_buff_size, void *data) { smx_comm_t comm; - /*If running in model-checking mode then intercept the communication action*/ - #ifdef HAVE_MC - if (_surf_do_model_check) - MC_trans_intercept_isend(rdv); - #endif + /*If running in model-checking mode then intercept the communication action */ +#ifdef HAVE_MC + if (_surf_do_model_check) + MC_trans_intercept_isend(rdv); +#endif /* Look for communication request matching our needs. If it is not found then create it and push it into the rendez-vous point */ comm = SIMIX_rdv_get_request(rdv, comm_recv); - if(!comm){ + if (!comm) { comm = SIMIX_communication_new(comm_send); SIMIX_rdv_push(rdv, comm); } @@ -497,16 +612,17 @@ smx_comm_t SIMIX_network_isend(smx_rdv_t rdv, double task_size, double rate, comm->src_buff = src_buff; comm->src_buff_size = src_buff_size; comm->data = data; - + SIMIX_communication_start(comm); return comm; } -smx_comm_t SIMIX_network_irecv(smx_rdv_t rdv, void *dst_buff, size_t *dst_buff_size) +smx_comm_t SIMIX_network_irecv(smx_rdv_t rdv, void *dst_buff, + size_t * dst_buff_size) { smx_comm_t comm; - /*If running in model-checking mode then intercept the communication action*/ + /*If running in model-checking mode then intercept the communication action */ #ifdef HAVE_MC if (_surf_do_model_check) MC_trans_intercept_irecv(rdv); @@ -516,7 +632,7 @@ smx_comm_t SIMIX_network_irecv(smx_rdv_t rdv, void *dst_buff, size_t *dst_buff_s */ comm = SIMIX_rdv_get_request(rdv, comm_send); - if(!comm){ + if (!comm) { comm = SIMIX_communication_new(comm_recv); SIMIX_rdv_push(rdv, comm); } @@ -525,7 +641,7 @@ smx_comm_t SIMIX_network_irecv(smx_rdv_t rdv, void *dst_buff, size_t *dst_buff_s comm->dst_proc = SIMIX_process_self(); comm->dst_buff = dst_buff; comm->dst_buff_size = dst_buff_size; - + SIMIX_communication_start(comm); return comm; } @@ -533,7 +649,7 @@ smx_comm_t SIMIX_network_irecv(smx_rdv_t rdv, void *dst_buff, size_t *dst_buff_s /** @brief blocks until the communication terminates or the timeout occurs */ XBT_INLINE void SIMIX_network_wait(smx_comm_t comm, double timeout) { - /*If running in model-checking mode then intercept the communication action*/ + /*If running in model-checking mode then intercept the communication action */ #ifdef HAVE_MC if (_surf_do_model_check) MC_trans_intercept_wait(comm); @@ -545,14 +661,14 @@ XBT_INLINE void SIMIX_network_wait(smx_comm_t comm, double timeout) /** @Returns whether the (asynchronous) communication is done yet or not */ XBT_INLINE int SIMIX_network_test(smx_comm_t comm) { - /*If running in model-checking mode then intercept the communication action*/ + /*If running in model-checking mode then intercept the communication action */ #ifdef HAVE_MC if (_surf_do_model_check) MC_trans_intercept_test(comm); #endif /* Copy data if the communication is done */ - if(comm->sem && !SIMIX_sem_would_block(comm->sem)){ + if (comm->sem && !SIMIX_sem_would_block(comm->sem)) { /* Copy network data */ SIMIX_network_copy_data(comm); return TRUE; @@ -566,26 +682,32 @@ XBT_INLINE int SIMIX_network_test(smx_comm_t comm) */ unsigned int SIMIX_network_waitany(xbt_dynar_t comms) { - xbt_dynar_t sems = xbt_dynar_new(sizeof(smx_sem_t),NULL); - unsigned int cursor, found_comm=-1; - smx_comm_t comm,comm_finished=NULL; + xbt_dynar_t sems = xbt_dynar_new(sizeof(smx_sem_t), NULL); + unsigned int cursor, found_comm = -1; + smx_comm_t comm, comm_finished = NULL; - /*If running in model-checking mode then intercept the communication action*/ + /*If running in model-checking mode then intercept the communication action */ #ifdef HAVE_MC if (_surf_do_model_check) MC_trans_intercept_waitany(comms); #endif - xbt_dynar_foreach(comms,cursor,comm) - xbt_dynar_push(sems,&(comm->sem)); + xbt_dynar_foreach(comms, cursor, comm) + xbt_dynar_push(sems, &(comm->sem)); DEBUG1("Waiting for the completion of communication set %p", comms); found_comm = SIMIX_sem_acquire_any(sems); xbt_dynar_free_container(&sems); - xbt_assert0(found_comm!=-1,"Cannot find which communication finished"); - xbt_dynar_get_cpy(comms,found_comm,&comm_finished); + xbt_assert0(found_comm != -1, + "Cannot find which communication finished"); + xbt_dynar_get_cpy(comms, found_comm, &comm_finished); + + DEBUG2("Communication %p of communication set %p finished", + comm_finished, comms); - /* Check for errors and cleanup the comm */ - SIMIX_communication_wait_for_completion(comm_finished,-1); + /* let the regular code deal with the communication end (errors checking and cleanup). + * A bit of useless work will be done, but that's good for source factorization */ + SIMIX_sem_release_forever(comm_finished->sem); + SIMIX_communication_wait_for_completion(comm_finished, -1); return found_comm; }