X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/7a4e264744a18d8953c3bd5e4a1df05f5d6088b9..354fdb4b40e5266ad8effe4af0699ea96d44c3fe:/src/simix/smx_network.c diff --git a/src/simix/smx_network.c b/src/simix/smx_network.c index 8890f81c67..91eef6bb9e 100644 --- a/src/simix/smx_network.c +++ b/src/simix/smx_network.c @@ -8,6 +8,13 @@ #include "private.h" #include "xbt/log.h" +#include "xbt/dict.h" + +/* Pimple to get an histogram of message sizes in the simulation */ +xbt_dict_t msg_sizes = NULL; + +XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix, + "Logging specific to SIMIX (network)"); /******************************************************************************/ /* Rendez-Vous Points */ @@ -18,88 +25,113 @@ * \param name The name of the rendez-vous point * \return The created rendez-vous point */ -smx_rvpoint_t SIMIX_rvpoint_create(char *name) +smx_rdv_t SIMIX_rdv_create(const char *name) { - smx_rvpoint_t rvp = xbt_new0(s_smx_rvpoint_t, 1); - rvp->name = xbt_strdup(name); - rvp->read = SIMIX_mutex_init(); - rvp->write = SIMIX_mutex_init(); - rvp->comm_mutex = SIMIX_mutex_init(); - rvp->comm_fifo = xbt_fifo_new(); - - return rvp; + smx_rdv_t rdv = xbt_new0(s_smx_rvpoint_t, 1); + rdv->name = name ? xbt_strdup(name) : NULL; + rdv->read = SIMIX_mutex_init(); + rdv->write = SIMIX_mutex_init(); + rdv->comm_fifo = xbt_fifo_new(); + + return rdv; } /** * \brief Destroy a rendez-vous point * \param name The rendez-vous point to destroy */ -void SIMIX_rvpoint_destroy(smx_rvpoint_t rvp) +void SIMIX_rdv_destroy(smx_rdv_t rdv) { - xbt_free(rvp->name); - SIMIX_mutex_destroy(rvp->read); - SIMIX_mutex_destroy(rvp->write); - SIMIX_mutex_destroy(rvp->comm_mutex); - xbt_fifo_free(rvp->comm_fifo); - xbt_free(rvp); + if(rdv->name) + xbt_free(rdv->name); + SIMIX_mutex_destroy(rdv->read); + SIMIX_mutex_destroy(rdv->write); + xbt_fifo_free(rdv->comm_fifo); + xbt_free(rdv); } /** * \brief Push a communication request into a rendez-vous point - * The communications request are dequeued by the two functions below - * \param rvp The rendez-vous point + * \param rdv The rendez-vous point * \param comm The communication request */ -static inline void SIMIX_rvpoint_push(smx_rvpoint_t rvp, smx_comm_t comm) +static inline void SIMIX_rdv_push(smx_rdv_t rdv, smx_comm_t comm) { - xbt_fifo_push(rvp->comm_fifo, comm); + xbt_fifo_push(rdv->comm_fifo, comm); + comm->rdv = rdv; } /** - * \brief Checks if there is a receive communication request queued in a rendez-vous - * \param rvp The rendez-vous with the queue - * \return The communication request if found, NULL otherwise. + * \brief Remove a communication request from a rendez-vous point + * \param rdv The rendez-vous point + * \param comm The communication request */ -smx_comm_t SIMIX_rvpoint_get_receiver(smx_rvpoint_t rvp) +static inline void SIMIX_rdv_remove(smx_rdv_t rdv, smx_comm_t comm) { - /* Get a communication request from the rendez-vous queue. If it is a receive - request then return it, otherwise put it again in the queue and return NULL - */ - smx_comm_t comm_head = xbt_fifo_shift(rvp->comm_fifo); - - if(comm_head != NULL && comm_head->dst_host != NULL) - return comm_head; + xbt_fifo_remove(rdv->comm_fifo, comm); + comm->rdv = NULL; +} - xbt_fifo_unshift(rvp->comm_fifo, comm_head); +/** + * \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)); + + if(comm && comm->type == type){ + DEBUG0("Communication request found!"); + xbt_fifo_shift(rdv->comm_fifo); + SIMIX_communication_use(comm); + comm->rdv = NULL; + return comm; + } + + DEBUG0("Communication request not found"); return NULL; } /** - * \brief Checks if there is a send communication request queued in a rendez-vous - * \param rvp The rendez-vous with the queue - * \return The communication request if found, NULL otherwise. + * \brief counts the number of communication requests of a given host pending + * on a rendez-vous point + * \param rdv The rendez-vous point + * \param host The host to be counted + * \return The number of comm request pending in the rdv */ -smx_comm_t SIMIX_rvpoint_get_sender(smx_rvpoint_t rvp) +int +SIMIX_rdv_get_count_waiting_comm(smx_rdv_t rdv, smx_host_t host) { - /* Get a communication request from the rendez-vous queue. If it is a send - request then return it, otherwise put it again in the queue and return NULL - */ - smx_comm_t comm_head = xbt_fifo_shift(rvp->comm_fifo); + smx_comm_t comm = NULL; + xbt_fifo_item_t item = NULL; + int count = 0; - if(comm_head != NULL && comm_head->src_host != NULL) - return comm_head; - - xbt_fifo_unshift(rvp->comm_fifo, comm_head); - return NULL; + xbt_fifo_foreach(rdv->comm_fifo, item, comm, smx_comm_t) { + if (comm->src_proc->smx_host == host) + count++; + } + + return count; } /** - * \brief Get the communication mutex of the rendez-vous point - * \param rvp The rendez-vous point + * \brief returns the communication at the head of the rendez-vous + * \param rdv The rendez-vous point + * \return The communication or NULL if empty */ -static inline smx_mutex_t SIMIX_rvpoint_get_comm_mutex(smx_rvpoint_t rvp) +XBT_INLINE smx_comm_t SIMIX_rdv_get_head(smx_rdv_t rdv) { - return rvp->comm_mutex; + 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; } /******************************************************************************/ @@ -108,22 +140,17 @@ static inline smx_mutex_t SIMIX_rvpoint_get_comm_mutex(smx_rvpoint_t rvp) /** * \brief Creates a new communication request - * \param sender The process starting the communication (by send) - * \param receiver The process receiving the communication (by recv) - * \return the communication request + * \param type The type of communication (comm_send, comm_recv) + * \return The new communication request */ -smx_comm_t SIMIX_communication_new(smx_host_t src_host, smx_host_t dst_host, - smx_rvpoint_t rdv) +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->cond = SIMIX_cond_init(); - - /* initialize them */ - comm->src_host = src_host; - comm->dst_host = dst_host; - comm->rdv = rdv; - + comm->type = type; + comm->sem = SIMIX_sem_init(0); + comm->refcount = 1; + return comm; } @@ -134,16 +161,30 @@ smx_comm_t SIMIX_communication_new(smx_host_t src_host, smx_host_t dst_host, void SIMIX_communication_destroy(smx_comm_t comm) { comm->refcount--; - if(comm->refcount == 0){ - if(comm->act != NULL) - SIMIX_action_destroy(comm->act); + if(comm->refcount > 0) + return; - if(comm->data != NULL) - xbt_free(comm->data); + if(comm->sem){ + SIMIX_sem_destroy(comm->sem); + comm->sem = NULL; + } + + if(comm->act){ + SIMIX_action_destroy(comm->act); + comm->act = NULL; + } - xbt_free(comm->cond); - xbt_free(comm); + 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); } /** @@ -161,153 +202,352 @@ static inline void SIMIX_communication_use(smx_comm_t comm) /** * \brief Start the simulation of a communication request - * \param comm The communicatino request + * \param comm The comm->rdv = NULL;communication request */ static inline void SIMIX_communication_start(smx_comm_t comm) { - comm->act = SIMIX_action_communicate(comm->src_host, comm->dst_host, NULL, - comm->data_size, comm->rate); + /* If both the sender and the receiver are already there, start the communication */ + 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->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 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); + } + + /* Add the communication as user data of the action */ + comm->act->data = comm; - SIMIX_register_action_to_condition(comm->act, comm->cond); - __SIMIX_cond_wait(comm->cond); - SIMIX_unregister_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); + } } - -/******************************************************************************/ -/* Synchronous Communication */ -/******************************************************************************/ -void SIMIX_network_send(smx_rvpoint_t rdv, void *data, size_t size, double timeout, double rate) +/** + * \brief Waits for communication completion and performs error checking + * \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 + */ +static inline void SIMIX_communication_wait_for_completion(smx_comm_t comm, double timeout) { - /*double start_time = SIMIX_get_clock();*/ - void *smx_net_data; - smx_host_t my_host = SIMIX_host_self(); - smx_comm_t comm; - smx_mutex_t rvp_comm_mutex = SIMIX_rvpoint_get_comm_mutex(rdv); - - /* Lock the rendez-vous point */ - SIMIX_mutex_lock(rvp_comm_mutex); - /* Copy the message to the network */ - /*FIXME here the MC should allocate the space from the network storage area */ - smx_net_data = xbt_malloc(size); - memcpy(smx_net_data, data, size); - - /* Check if there is already a receive waiting in the rendez-vous point */ - if((comm = SIMIX_rvpoint_get_receiver(rdv)) != NULL){ - comm->src_host = my_host; - comm->data = smx_net_data; - comm->data_size = size; - comm->rate = rate; - SIMIX_communication_use(comm); + smx_action_t act_sleep = NULL; + int src_timeout = 0; + int dst_timeout = 0; - /* Unlock the rendez-vous point and start the communication action.*/ - /* FIXME: if the semantic is non blocking, it shouldn't wait on the condition here */ - SIMIX_mutex_unlock(rvp_comm_mutex); - SIMIX_communication_start(comm); - - /* Nobody is at the rendez-vous point, so push the comm action into it */ - }else{ - comm = SIMIX_communication_new(my_host, NULL, rdv); - comm->data = smx_net_data; - comm->data_size = size; - comm->rate = rate; - SIMIX_communication_use(comm); - SIMIX_rvpoint_push(rdv, comm); + 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)); + 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); + } - /* Wait for communication completion */ - /* FIXME: if the semantic is non blocking, it shouldn't wait on the condition here */ - /* FIXME: add timeout checking stuff */ - SIMIX_mutex_unlock (rvp_comm_mutex); - __SIMIX_cond_wait(comm->cond); + /* 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); + + THROW1(timeout_error, 0, "Communication timeouted because of %s",src_timeout?"the source":"the destination"); } - /* Check for errors */ - if (SIMIX_host_get_state(comm->dst_host) == 0){ - THROW1(host_error, 0, "Destination host %s failed", comm->dst_host->name); - }else if(SIMIX_action_get_state(comm->act) == SURF_ACTION_FAILED){ + 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"); } - SIMIX_communication_destroy(comm); - return; } -void SIMIX_network_recv(smx_rvpoint_t rdv, void **data, size_t *size, double timeout) +/** + * \brief Cancels a communication + * \brief comm The communication to cancel + */ +XBT_INLINE void SIMIX_communication_cancel(smx_comm_t comm) { - /*double start_time = SIMIX_get_clock();*/ - smx_comm_t comm; - smx_host_t my_host = SIMIX_host_self(); - smx_mutex_t rvp_comm_mutex = SIMIX_rvpoint_get_comm_mutex(rdv); + if (comm->act) + SIMIX_action_cancel(comm->act); +} - /* Lock the rendez-vous point */ - SIMIX_mutex_lock(rvp_comm_mutex); - - /* Check if there is already a send waiting in the rendez-vous point */ - if((comm = SIMIX_rvpoint_get_sender(rdv)) != NULL){ - comm->dst_host = my_host; - comm->dest_buff = data; - SIMIX_communication_use(comm); +/** + * \brief get the amount remaining from the communication + * \param comm The communication + */ +XBT_INLINE double SIMIX_communication_get_remains(smx_comm_t comm) +{ + return SIMIX_action_get_remains(comm->act); +} - /* Unlock the rendez-vous point and start the communication action.*/ - /* FIXME: if the semantic is non blocking, it shouldn't wait on the condition here */ - SIMIX_mutex_unlock(rvp_comm_mutex); - SIMIX_communication_start(comm); +/******************************************************************************/ +/* SIMIX_network_copy_data callbacks */ +/******************************************************************************/ +static void (*SIMIX_network_copy_data_callback)(smx_comm_t, size_t) = &SIMIX_network_copy_pointer_callback; - /* Nobody is at the rendez-vous point, so push the comm action into it */ - }else{ - comm = SIMIX_communication_new(NULL, my_host, rdv); - comm->dest_buff = data; - SIMIX_communication_use(comm); - SIMIX_rvpoint_push(rdv, comm); +void SIMIX_network_set_copy_data_callback(void (*callback)(smx_comm_t, size_t)) { + SIMIX_network_copy_data_callback = callback; +} - /* Wait for communication completion */ - /* FIXME: if the semantic is non blocking, it shouldn't wait on the condition here */ - /* FIXME: add timeout checking stuff*/ - SIMIX_mutex_unlock (rvp_comm_mutex); - __SIMIX_cond_wait(comm->cond); - } +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; +} - /* Check for errors */ - if (SIMIX_host_get_state(comm->src_host) == 0){ - THROW1(host_error, 0, "Source host %s failed", comm->src_host->name); - }else if(SIMIX_action_get_state(comm->act) == SURF_ACTION_FAILED){ - THROW0(network_error, 0, "Link failure"); - } +void SIMIX_network_copy_buffer_callback(smx_comm_t comm, size_t buff_size) { + memcpy(comm->dst_buff, comm->src_buff, buff_size); +} - /* We are OK, let's copy the message to receiver's buffer */ - *size = *size < comm->data_size ? *size : comm->data_size; - memcpy(*data, comm->data, *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) +{ + /* If there is no data to be copy then return */ + if(!comm->src_buff || !comm->dst_buff) + return; - SIMIX_communication_destroy(comm); - return; + /* Copy at most dst_buff_size bytes of the message to receiver's buffer */ + size_t buff_size = comm->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 */ + if (comm->dst_buff_size) + *comm->dst_buff_size = buff_size; + + if(buff_size == 0) + 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); + (*SIMIX_network_copy_data_callback)(comm, buff_size); + + /* 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); + 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"); + INFO1("Output message sizes to %s",filename); + xbt_assert1(out,"Cannot open file %s",filename); + uintptr_t key,data; + xbt_dict_cursor_t cursor; + xbt_dict_foreach(msg_sizes,cursor,key,data) { + fprintf(out,"%ld %ld\n",key,data); + } + fclose(out); +} + +/** + * \brief Return the user data associated to the communication + * \param comm The communication + * \return the user data + */ +XBT_INLINE void *SIMIX_communication_get_data(smx_comm_t comm) +{ + return comm->data; } /******************************************************************************/ -/* Asynchronous Communication */ +/* Synchronous Communication */ /******************************************************************************/ +/** + * \brief Put a send communication request in a rendez-vous point and waits for + * its completion (blocking) + * \param rdv The rendez-vous point + * \param task_size The size of the communication action (for surf simulation) + * \param rate The rate of the communication action (for surf) + * \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 (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 + */ +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); +} -/* -void SIMIX_network_wait(smx_action_t comm, double 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) { - if (timeout > 0) - SIMIX_cond_wait_timeout(rvp_cond, rvp_comm_mutex, timeout - start_time); - else - SIMIX_cond_wait(rvp_cond, rvp_comm_mutex); + *comm_ref = 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 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){ + comm = SIMIX_communication_new(comm_send); + SIMIX_rdv_push(rdv, comm); + } + + /* Setup the communication request */ + comm->src_proc = SIMIX_process_self(); + comm->task_size = task_size; + comm->rate = 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 comm; -XBT_PUBLIC(int) SIMIX_network_test(smx_action_t comm) -{ - if(SIMIX_action_get_state (comm) == SURF_ACTION_DONE){ - memcpy(comm->data - - return SIMIX_action_get_state (comm) == SURF_ACTION_DONE ? TRUE : FALSE; -}*/ + /* 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){ + comm = SIMIX_communication_new(comm_recv); + SIMIX_rdv_push(rdv, 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) { + /* Wait for communication completion */ + SIMIX_communication_wait_for_completion(comm, timeout); +} + +/** @Returns whether the (asynchronous) communication is done yet or not */ +XBT_INLINE int SIMIX_network_test(smx_comm_t comm) { + return comm->sem?SIMIX_sem_would_block(comm->sem):0; +} + +/** @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) { + 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_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_assert0(found_comm!=-1,"Cannot find which communication finished"); + xbt_dynar_get_cpy(comms,found_comm,&comm_finished); + DEBUG1("Communication %p complete! Let's check for errors", comm_finished); + + /* Make sure that everyone sleeping on that semaphore is awake, + * and that nobody will ever block on it */ + SIMIX_sem_release_forever(comm_finished->sem); + + /* Check for errors */ + if(!SIMIX_host_get_state(SIMIX_host_self())){ + if(comm_finished->rdv) + SIMIX_rdv_remove(comm_finished->rdv, comm_finished); + SIMIX_communication_destroy(comm_finished); + THROW0(host_error, 0, "Host failed"); + } else if (SIMIX_action_get_state(comm_finished->act) == SURF_ACTION_FAILED){ + SIMIX_communication_destroy(comm_finished); + THROW0(network_error, 0, "Link failure"); + } + SIMIX_communication_destroy(comm_finished); + + return found_comm; +}