X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/aeb67d26a90a4fbc05a2ee596c8f627bb7aa615b..b38d5c15755a98cc7ef99a0a9c4936357dbedce3:/src/simix/smx_network.c diff --git a/src/simix/smx_network.c b/src/simix/smx_network.c index f9ac690d6f..c8c5ac9363 100644 --- a/src/simix/smx_network.c +++ b/src/simix/smx_network.c @@ -1,13 +1,19 @@ -/* $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_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix, + "Logging specific to SIMIX (network)"); /******************************************************************************/ /* Rendez-Vous Points */ @@ -20,81 +26,128 @@ */ smx_rdv_t SIMIX_rdv_create(const char *name) { - smx_rdv_t rvp = xbt_new0(s_smx_rvpoint_t, 1); - rvp->name = name ? xbt_strdup(name) : NULL; - rvp->read = SIMIX_mutex_init(); - rvp->write = SIMIX_mutex_init(); - rvp->comm_fifo = xbt_fifo_new(); + 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 rvp; + return rdv; } /** * \brief Destroy a rendez-vous point * \param name The rendez-vous point to destroy */ -void SIMIX_rdv_destroy(smx_rdv_t rvp) +void SIMIX_rdv_destroy(smx_rdv_t rdv) { - if(rvp->name) - xbt_free(rvp->name); - SIMIX_mutex_destroy(rvp->read); - SIMIX_mutex_destroy(rvp->write); - 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_rdv_push(smx_rdv_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 Remove a communication request from a rendez-vous point + * \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) +{ + 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 rvp The rendez-vous with the queue - * \param look_for_src boolean. True: we are receiver looking for sender; False: other way round - * \return The communication request if found, or a newly created one otherwise. + * \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 rvp, int (filter)(smx_comm_t, void*), void *arg) { - smx_comm_t comm; - xbt_fifo_item_t item; - - /* Traverse the rendez-vous queue looking for a comm request matching the - filter conditions. If found return it and remove it from the list. */ - xbt_fifo_foreach(rvp->comm_fifo, item, comm, smx_comm_t) { - if(filter(comm, arg)){ - SIMIX_communication_use(comm); - xbt_fifo_remove_item(rvp->comm_fifo, item); - return comm; - } +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; } - /* no relevant request found. Return NULL */ + DEBUG0("Communication request not found"); return NULL; } +/** + * \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 + */ +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; + int count = 0; + + xbt_fifo_foreach(rdv->comm_fifo, item, comm, smx_comm_t) { + if (comm->src_proc->smx_host == host) + count++; + } + + return count; +} + +/** + * \brief returns the communication at the head of the rendez-vous + * \param rdv The rendez-vous point + * \return The communication or NULL if empty + */ +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 */ /******************************************************************************/ /** * \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_comm_type_t type, smx_rdv_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->type = type; - comm->cond = SIMIX_cond_init(); - comm->rdv = rdv; + comm->sem = SIMIX_sem_init(0); comm->refcount = 1; return comm; @@ -107,13 +160,30 @@ smx_comm_t SIMIX_communication_new(smx_comm_type_t type, smx_rdv_t rdv) 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; - xbt_free(comm->cond); - xbt_free(comm); + 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); } /** @@ -136,13 +206,29 @@ static inline void SIMIX_communication_use(smx_comm_t comm) static 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_host != NULL && comm->dst_host != NULL){ - comm->act = SIMIX_action_communicate(comm->src_host, comm->dst_host, NULL, + 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); - /* Add the communication as user data into the action, so it can be reached from it later */ - comm->act->data = comm; +#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_suspend(comm->act); + } - SIMIX_register_action_to_condition(comm->act, comm->cond); + /* Add the communication as user data of the action */ + comm->act->data = comm; + + /* 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); } } @@ -152,192 +238,362 @@ static inline void SIMIX_communication_start(smx_comm_t comm) * \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) { - 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 it's a timeout then cancel the communication and signal the other peer */ - if(e.category == timeout_error) - SIMIX_action_cancel(comm->act); - SIMIX_cond_signal(comm->cond); - 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); } - SIMIX_unregister_action_to_condition(comm->act, 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"); + } + + 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 */ - if (SIMIX_host_get_state(comm->dst_host) == 0){ - THROW1(host_error, 0, "Destination host %s failed", comm->dst_host->name); - } else if (SIMIX_host_get_state(comm->src_host) == 0){ - THROW1(host_error, 0, "Source host %s failed", comm->src_host->name); + /* 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); } /** - * \brief Copy the communication data from the sender's buffer to the receiver's one - * \param comm The communication + * \brief Cancels a communication + * \brief comm The communication to cancel */ -void SIMIX_network_copy_data(smx_comm_t comm) +XBT_INLINE void SIMIX_communication_cancel(smx_comm_t comm) { - /* Copy the minimum between the size of the sender's message and the size of the - receiver's buffer */ - *comm->dest_buff_size = *comm->dest_buff_size < comm->data_size ? - *comm->dest_buff_size : comm->data_size; - - memcpy(comm->dest_buff, comm->data, *comm->dest_buff_size); + if (comm->act) + SIMIX_action_cancel(comm->act); } /** - * \brief Checks if a communication is a send request + * \brief get the amount remaining from the communication * \param comm The communication - * \return Boolean value */ -int SIMIX_communication_isSend(smx_comm_t comm) +XBT_INLINE double SIMIX_communication_get_remains(smx_comm_t comm) { - return comm->type == comm_send ? TRUE : FALSE; + return SIMIX_action_get_remains(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 Checks if a communication is a recv request + * \brief Copy the communication data from the sender's buffer to the receiver's one * \param comm The communication - * \return Boolean value */ -int SIMIX_communication_isRecv(smx_comm_t comm) +void SIMIX_network_copy_data(smx_comm_t comm) { - return comm->type == comm_recv ? TRUE : FALSE; -} + size_t buff_size = comm->src_buff_size; -/* FIXME: move to some other place */ -int comm_filter_get(smx_comm_t comm, void *arg) -{ - if(comm->type == comm_send){ - if(arg && comm->src_host != (smx_host_t)arg) - return FALSE; - else - return TRUE; - }else{ - return FALSE; + /* If there is no data to be copy then return */ + 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); + + /* 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)); + + /* 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; + (*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)*/ + 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); + 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; + xbt_dict_cursor_t cursor; + xbt_dict_foreach(msg_sizes,cursor,key,data) { + fprintf(out,"%zu %zu\n",key,data); + } + fclose(out); +} -int comm_filter_put(smx_comm_t comm, void *arg) +/** + * \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->type == comm_recv ? TRUE : FALSE; + return comm->data; } + /******************************************************************************/ /* Synchronous Communication */ /******************************************************************************/ -/* Throws: +/** + * \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); +} + +/** + * \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 */ -void SIMIX_network_send(smx_rdv_t rdv, double task_size, double rate, - double timeout, void *data, size_t data_size, - int (filter)(smx_comm_t, void *), void *arg) +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); +} + +/******************************************************************************/ +/* 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; + mc_transition_t trans=NULL; - /* Look for communication request matching our needs. + if (_surf_do_model_check) { + /* Let's intercept the communication and control it from the model-checker */ + trans = MC_create_transition(mc_isend, SIMIX_process_self(), rdv, NULL); + SIMIX_process_yield(); + } + + /* 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, filter, arg); + comm = SIMIX_rdv_get_request(rdv, comm_recv); - if(comm == NULL){ - comm = SIMIX_communication_new(comm_send, rdv); + if(!comm){ + comm = SIMIX_communication_new(comm_send); SIMIX_rdv_push(rdv, comm); } /* Setup the communication request */ - comm->src_host = SIMIX_host_self(); + 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; - comm->data_size = data_size; + /* Associate the simix communication to the mc transition */ + if (_surf_do_model_check) + MC_transition_set_comm(trans, comm); + SIMIX_communication_start(comm); - - /* 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_communication_wait_for_completion(comm, timeout); - - SIMIX_communication_destroy(comm); + return comm; } -/* 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 *data, - size_t *data_size, int (filter)(smx_comm_t, void *), void *arg) -{ +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. + mc_transition_t trans=NULL; + + if (_surf_do_model_check) { + /* Let's intercept the communication and control it from the model-checker */ + trans = MC_create_transition(mc_irecv, SIMIX_process_self(), rdv, NULL); + SIMIX_process_yield(); + } + + /* 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, filter, arg); + comm = SIMIX_rdv_get_request(rdv, comm_send); - if(comm == NULL){ - comm = SIMIX_communication_new(comm_send, rdv); + if(!comm){ + comm = SIMIX_communication_new(comm_recv); SIMIX_rdv_push(rdv, comm); } /* Setup communication request */ - comm->dst_host = SIMIX_host_self(); - comm->dest_buff = data; - comm->dest_buff_size = data_size; - + comm->dst_proc = SIMIX_process_self(); + comm->dst_buff = dst_buff; + comm->dst_buff_size = dst_buff_size; + + /* Associate the simix communication to the mc transition */ + if (_surf_do_model_check) + MC_transition_set_comm(trans, comm); + 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 (_surf_do_model_check) { + /* Let's intercept the communication and control it from the model-checker */ + MC_create_transition(mc_wait, SIMIX_process_self(), comm->rdv, comm); + SIMIX_process_yield(); + } /* 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_communication_wait_for_completion(comm, timeout); - - SIMIX_communication_destroy(comm); } -/******************************************************************************/ -/* Asynchronous Communication */ -/******************************************************************************/ - -/* -void SIMIX_network_wait(smx_action_t comm, double timeout) -{ - if (timeout > 0) - SIMIX_cond_wait_timeout(rvp_cond, rvp_comm_mutex, timeout - start_time); - else - SIMIX_cond_wait(rvp_cond, rvp_comm_mutex); +/** @Returns whether the (asynchronous) communication is done yet or not */ +XBT_INLINE int SIMIX_network_test(smx_comm_t comm) { + if (_surf_do_model_check) { + /* Let's intercept the communication and control it from the model-checker */ + MC_create_transition(mc_test, SIMIX_process_self(), comm->rdv, comm); + SIMIX_process_yield(); + } + /* 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; } +/** @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; + + if (_surf_do_model_check) { + /* Let's intercept the communication and control it from the model-checker */ + MC_create_transition(mc_waitany, SIMIX_process_self(), NULL, NULL); + SIMIX_process_yield(); + } + + xbt_dynar_foreach(comms,cursor,comm) + xbt_dynar_push(sems,&(comm->sem)); -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; -}*/ - - - - - + 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); + /* Check for errors and cleanup the comm */ + SIMIX_communication_wait_for_completion(comm_finished,-1); + return found_comm; +}