X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/44723299849907b8923bfac8f9b111bcc07f0987..e5487aa8732f87d753fd97a8968006950717d894:/src/simix/smx_network.c diff --git a/src/simix/smx_network.c b/src/simix/smx_network.c index cdab3e1914..03782b4432 100644 --- a/src/simix/smx_network.c +++ b/src/simix/smx_network.c @@ -1,13 +1,17 @@ -/* $Id$ */ - -/* Copyright (c) 2009 Cristian Rosa. - All rights reserved. */ +/* Copyright (c) 2009, 2010. The SimGrid Team. + * All rights reserved. */ /* 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 "private.h" #include "xbt/log.h" +#include "mc/mc.h" +#include "xbt/dict.h" + +/* Pimple to get an histogram of message sizes in the simulation */ +xbt_dict_t msg_sizes = NULL; +xbt_dict_t latency_limited_dict = NULL; XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix, "Logging specific to SIMIX (network)"); @@ -51,7 +55,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; @@ -62,7 +66,7 @@ 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; @@ -73,8 +77,7 @@ static inline void SIMIX_rdv_remove(smx_rdv_t rdv, smx_comm_t comm) * \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 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)); @@ -82,11 +85,10 @@ smx_comm_t SIMIX_rdv_get_request(smx_rdv_t rdv, smx_comm_type_t type) DEBUG0("Communication request found!"); xbt_fifo_shift(rdv->comm_fifo); SIMIX_communication_use(comm); - comm->rdv = NULL; + comm->rdv = NULL; return comm; } - /* no relevant request found. Return NULL */ DEBUG0("Communication request not found"); return NULL; } @@ -118,11 +120,19 @@ SIMIX_rdv_get_count_waiting_comm(smx_rdv_t rdv, smx_host_t host) * \param rdv The rendez-vous point * \return The communication or NULL if empty */ -smx_comm_t SIMIX_rdv_get_head(smx_rdv_t rdv) +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)); } +/** @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; +} +/** @brief gets API-related data from the rendez-vous point */ +XBT_INLINE void *SIMIX_rdv_get_data(smx_rdv_t rdv) { + return rdv->data; +} /******************************************************************************/ /* Communication Requests */ @@ -138,7 +148,7 @@ smx_comm_t SIMIX_communication_new(smx_comm_type_t type) /* alloc structures */ smx_comm_t comm = xbt_new0(s_smx_comm_t, 1); comm->type = type; - comm->cond = SIMIX_cond_init(); + comm->sem = SIMIX_sem_init(0); comm->refcount = 1; return comm; @@ -150,20 +160,40 @@ 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); + + //save is latency limited flag to use afterwards + if (latency_limited_dict == NULL) + latency_limited_dict = xbt_dict_new(); + 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)); + comm->refcount--; if(comm->refcount > 0) return; - if(comm->cond){ - SIMIX_cond_destroy(comm->cond); - comm->cond = NULL; + if(comm->sem){ + SIMIX_sem_destroy(comm->sem); + comm->sem = NULL; } if(comm->act){ SIMIX_action_destroy(comm->act); comm->act = NULL; } - + + 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; + } + + + xbt_free(comm); } @@ -175,16 +205,16 @@ 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++; } /** * \brief Start the simulation of a communication request - * \param comm The comm->rdv = NULL;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){ @@ -192,119 +222,241 @@ static inline void SIMIX_communication_start(smx_comm_t comm) 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); +#endif /* If any of the process is suspend, create the action but stop its execution, - it will be restart when the sender process resume */ + it will be restarted when the sender process resume */ if(SIMIX_process_is_suspended(comm->src_proc) || SIMIX_process_is_suspended(comm->dst_proc)) { - SIMIX_action_set_priority(comm->act, 0); + SIMIX_action_suspend(comm->act); } /* Add the communication as user data of the action */ comm->act->data = comm; - - SIMIX_register_action_to_condition(comm->act, comm->cond); + + /* The semaphore will only get signaled once, but since the first unlocked guy will + * release_forever() the semaphore, that will unlock the second (and any other) + * communication partner */ + SIMIX_register_action_to_semaphore(comm->act, comm->sem); } } /** - * \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 * * Throws: - * - host_error if peer failed - * - timeout_error if communication reached the timeout specified - * - network_error if network failed or peer issued a timeout + * - host_error if local peer failed + * - 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) { - xbt_ex_t e; + 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){ - TRY{ - SIMIX_cond_wait_timeout(comm->cond, NULL, timeout); - } - CATCH(e){ - /* If there is a timeout then cancel the communication if it is running or - remove it from the rendez-vous otherwise. Then signal the other peer, - destroy the communication and retrow the exception. */ - if(e.category == timeout_error){ - DEBUG1("Communication timeout! %p", comm); - if(comm->act && SIMIX_action_get_state(comm->act) == SURF_ACTION_RUNNING) - SIMIX_communication_cancel(comm); - else - SIMIX_rdv_remove(comm->rdv, comm); - - SIMIX_cond_signal(comm->cond); - SIMIX_communication_destroy(comm); - } - RETHROW; - } - }else{ - SIMIX_cond_wait(comm->cond, NULL); + 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)); + SIMIX_register_action_to_semaphore(act_sleep, comm->sem); + SIMIX_process_self()->waiting_action = act_sleep; + SIMIX_sem_block_onto(comm->sem); + SIMIX_process_self()->waiting_action = NULL; + SIMIX_unregister_action_to_semaphore(act_sleep, comm->sem); + } else { + SIMIX_sem_acquire(comm->sem); } - DEBUG1("Communication %p complete! Let's check for errors", comm); - - /* Check for errors other than timeouts (they are catched above) */ - if(!SIMIX_host_get_state(SIMIX_host_self())){ - 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){ + /* 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) + SIMIX_communication_cancel(comm); + else if (comm->rdv) + SIMIX_rdv_remove(comm->rdv, 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); SIMIX_communication_destroy(comm); - THROW0(network_error, 0, "Link failure"); + + THROW1(timeout_error, 0, "Communication timeouted because of %s",src_timeout?"the source":"the destination"); } + + DEBUG1("Communication %p complete!", comm); + SIMIX_communication_cleanup(comm); } /** * \brief Cancels a communication * \brief comm The communication to cancel */ -void SIMIX_communication_cancel(smx_comm_t comm) +XBT_INLINE void SIMIX_communication_cancel(smx_comm_t comm) { - SIMIX_action_cancel(comm->act); + if (comm->act) + SIMIX_action_cancel(comm->act); } /** * \brief get the amount remaining from the communication * \param comm The communication */ -double SIMIX_communication_get_remains(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); } +/** + * \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); +} + +/******************************************************************************/ +/* SIMIX_network_copy_data callbacks */ +/******************************************************************************/ +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)) { + 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_buffer_callback(smx_comm_t comm, size_t buff_size) { + memcpy(comm->dst_buff, comm->src_buff, buff_size); +} + /** * \brief Copy the communication data from the sender's buffer to the receiver's one * \param comm The communication */ 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) + if(!comm->src_buff || !comm->dst_buff || comm->copied == 1) return; - - size_t src_buff_size = comm->src_buff_size; - size_t dst_buff_size = *comm->dst_buff_size; - + + 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); + /* Copy at most dst_buff_size bytes of the message to receiver's buffer */ - dst_buff_size = MIN(dst_buff_size, src_buff_size); + if (comm->dst_buff_size) + buff_size = MIN(buff_size,*(comm->dst_buff_size)); /* Update the receiver's buffer size to the copied amount */ - *comm->dst_buff_size = dst_buff_size; + if (comm->dst_buff_size) + *comm->dst_buff_size = buff_size; - if(dst_buff_size == 0) + if(buff_size == 0) return; + (*SIMIX_network_copy_data_callback)(comm, buff_size); - memcpy(comm->dst_buff, comm->src_buff, dst_buff_size); - - DEBUG4("Copying comm %p data from %s -> %s (%zu bytes)", - comm, comm->src_proc->smx_host->name, comm->dst_proc->smx_host->name, - dst_buff_size); + /* Set the copied flag so we copy data only once */ + /* (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(); + casted_size = comm->task_size; + amount = xbt_dicti_get(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) { + uintptr_t key = 0; + uintptr_t data = 0; + xbt_dict_cursor_t cursor; + 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); } /** @@ -312,11 +464,28 @@ void SIMIX_network_copy_data(smx_comm_t comm) * \param comm The communication * \return the user data */ -void *SIMIX_communication_get_data(smx_comm_t comm) +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 */ /******************************************************************************/ @@ -329,20 +498,55 @@ void *SIMIX_communication_get_data(smx_comm_t comm) * \param timeout The timeout used for the waiting the completion * \param src_buff The source buffer containing the message to be sent * \param src_buff_size The size of the source buffer - * \param comm_ref The communication object used for the send + * \param comm_ref The communication object used for the send (useful if someone else wants to cancel this communication afterward) * \param data User data associated to the communication object * Throws: * - host_error if peer failed * - timeout_error if communication reached the timeout specified * - network_error if network failed or peer issued a timeout */ -void SIMIX_network_send(smx_rdv_t rdv, double task_size, double rate, +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); +} + +/** + * \brief Put a receive communication request in a rendez-vous point and waits + * for its completion (blocking) + * \param rdv The rendez-vous point + * \param timeout The timeout used for the waiting the completion + * \param dst_buff The destination buffer to copy the received message + * \param src_buff_size The size of the destination buffer + * \param comm_ref The communication object used for the send (useful if someone else wants to cancel this communication afterward) + * Throws: + * - host_error if peer failed + * - 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) +{ + *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 comm; - - /* Look for communication request matching our needs. + + /*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); @@ -351,9 +555,6 @@ void SIMIX_network_send(smx_rdv_t rdv, double task_size, double rate, SIMIX_rdv_push(rdv, comm); } - /* Update the communication reference with the comm to be used */ - *comm_ref = comm; - /* Setup the communication request */ comm->src_proc = SIMIX_process_self(); comm->task_size = task_size; @@ -361,35 +562,23 @@ void SIMIX_network_send(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); - - /* Wait for communication completion */ - SIMIX_communication_wait_for_completion(comm, timeout); - - SIMIX_communication_destroy(comm); + return comm; } -/** - * \brief Put a receive communication request in a rendez-vous point and waits - * for its completion (blocking) - * \param rdv The rendez-vous point - * \param timeout The timeout used for the waiting the completion - * \param dst_buff The destination buffer to copy the received message - * \param src_buff_size The size of the destination buffer - * \param comm_ref The communication object used for the send - * Throws: - * - host_error if peer failed - * - timeout_error if communication reached the timeout specified - * - network_error if network failed or peer issued a timeout - */ -void SIMIX_network_recv(smx_rdv_t rdv, double timeout, void *dst_buff, - size_t *dst_buff_size, smx_comm_t *comm_ref) +smx_comm_t SIMIX_network_irecv(smx_rdv_t rdv, void *dst_buff, size_t *dst_buff_size) { smx_comm_t comm; - /* Look for communication request matching our needs. - If it is not found then create it and push it into the rendez-vous point */ + /*If running in model-checking mode then intercept the communication action*/ +#ifdef HAVE_MC + if (_surf_do_model_check) + MC_trans_intercept_irecv(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_send); if(!comm){ @@ -397,40 +586,75 @@ void SIMIX_network_recv(smx_rdv_t rdv, double timeout, void *dst_buff, SIMIX_rdv_push(rdv, comm); } - /* Update the communication reference with the comm to be used */ - *comm_ref = comm; - /* Setup communication request */ comm->dst_proc = SIMIX_process_self(); comm->dst_buff = dst_buff; comm->dst_buff_size = dst_buff_size; - + SIMIX_communication_start(comm); + return comm; +} +/** @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*/ +#ifdef HAVE_MC + if (_surf_do_model_check) + MC_trans_intercept_wait(comm); +#endif /* Wait for communication completion */ SIMIX_communication_wait_for_completion(comm, timeout); - - SIMIX_communication_destroy(comm); } -/******************************************************************************/ -/* Asynchronous Communication */ -/******************************************************************************/ - -/* -void SIMIX_network_wait(smx_action_t comm, double timeout) +/** @Returns whether the (asynchronous) communication is done yet or not */ +XBT_INLINE int SIMIX_network_test(smx_comm_t comm) { - TO BE IMPLEMENTED + /*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)){ + /* Copy network data */ + SIMIX_network_copy_data(comm); + return TRUE; + } + return FALSE; } -XBT_PUBLIC(int) SIMIX_network_test(smx_action_t comm) +/** @brief wait for the completion of any communication of a set + * + * @Returns the rank in the dynar of communication which finished; destroy it after identifying which one it is + */ +unsigned int SIMIX_network_waitany(xbt_dynar_t comms) { - TO BE IMPLEMENTED -}*/ - - - - - - - + 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*/ +#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)); + + 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); + + DEBUG2("Communication %p of communication set %p finished", comm_finished, comms); + + /* 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; +}