X-Git-Url: http://info.iut-bm.univ-fcomte.fr/pub/gitweb/simgrid.git/blobdiff_plain/f9143fc186eb0d357c8e15eae823f92c64b63347..6b28daccf7142bd4b00e9eee1cd63daa90ae4ad9:/src/simix/smx_network.c diff --git a/src/simix/smx_network.c b/src/simix/smx_network.c index 4e2a96ca0f..29e1bcf118 100644 --- a/src/simix/smx_network.c +++ b/src/simix/smx_network.c @@ -1,20 +1,26 @@ -/* $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; +#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 @@ -23,85 +29,134 @@ XBT_LOG_NEW_DEFAULT_SUBCATEGORY(simix_network, simix, */ 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 XBT_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 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 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); - DEBUG1("Communication request found! %p", comm); - 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 - */ -smx_comm_t SIMIX_communication_new(smx_comm_type_t type, smx_rdv_t rdv) + * \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 */ 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; } @@ -111,14 +166,49 @@ smx_comm_t SIMIX_communication_new(smx_comm_type_t type, smx_rdv_t rdv) */ 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->act != NULL) - SIMIX_action_destroy(comm->act); + if (comm->refcount > 0) + return; + + 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); } /** @@ -129,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++; } @@ -138,229 +228,486 @@ 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_host != NULL && comm->dst_host != NULL){ + if (comm->src_proc && comm->dst_proc) { DEBUG1("Starting communication %p", comm); - comm->act = SIMIX_action_communicate(comm->src_host, comm->dst_host, NULL, + 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 */ +#ifdef HAVE_TRACING + 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)) { + 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); - }else{ - DEBUG1("Communication %p cannot be started, peer missing", 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); } } /** - * \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 it's a timeout then cancel the communication and signal the other peer */ - if(e.category == timeout_error){ - DEBUG1("Communication timeout! %p", comm); - if(SIMIX_action_get_state(comm->act) == SURF_ACTION_RUNNING) - SIMIX_action_cancel(comm->act); - SIMIX_cond_signal(comm->cond); - /* FIXME: remove from the rvp if the communication didn't started */ - 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 */ - if (SIMIX_host_get_state(comm->dst_host) == 0){ - SIMIX_communication_destroy(comm); - THROW1(host_error, 0, "Destination host %s failed", comm->dst_host->name); - } else if (SIMIX_host_get_state(comm->src_host) == 0){ + /* 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(host_error, 0, "Source host %s failed", comm->src_host->name); - } else if (SIMIX_action_get_state(comm->act) == SURF_ACTION_FAILED){ - 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"); } - SIMIX_unregister_action_to_condition(comm->act, comm->cond); + DEBUG1("Communication %p complete!", comm); + SIMIX_communication_cleanup(comm); } /** - * \brief Copy the communication data from the sender's buffer to the receiver's one + * \brief Cancels a communication + * \brief comm The communication to cancel + */ +XBT_INLINE void SIMIX_communication_cancel(smx_comm_t comm) +{ + if (comm->act) + SIMIX_action_cancel(comm->act); +} + +/** + * \brief get the amount remaining from the communication * \param comm The communication */ -void SIMIX_network_copy_data(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; + +void +SIMIX_network_set_copy_data_callback(void (*callback) (smx_comm_t, size_t)) { - /* 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; + SIMIX_network_copy_data_callback = callback; +} - DEBUG4("Copying comm %p data from %s -> %s (%zu bytes)", comm, - comm->src_host->name, comm->dst_host->name, *comm->dest_buff_size); +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; +} - memcpy(comm->dest_buff, comm->data, *comm->dest_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); } /** - * \brief Checks if a communication is a send 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_isSend(smx_comm_t comm) +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) + 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(); + 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) { - return comm->type == comm_send ? TRUE : FALSE; + 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); } /** - * \brief Checks if a communication is a recv request + * \brief Return the user data associated to the communication * \param comm The communication - * \return Boolean value + * \return the user data */ -int SIMIX_communication_isRecv(smx_comm_t comm) +XBT_INLINE void *SIMIX_communication_get_data(smx_comm_t comm) { - return comm->type == comm_recv ? TRUE : FALSE; + return comm->data; } -/* FIXME: move to some other place */ -int comm_filter_get(smx_comm_t comm, void *arg) +XBT_PUBLIC(void *) SIMIX_communication_get_src_buf(smx_comm_t comm) { - if(comm->type == comm_send){ - if(arg && comm->src_host != (smx_host_t)arg) - return FALSE; - else - return TRUE; - }else{ - return FALSE; - } + return comm->src_buff; } -int comm_filter_put(smx_comm_t comm, void *arg) +XBT_PUBLIC(void *) SIMIX_communication_get_dst_buf(smx_comm_t comm) { - return comm->type == comm_recv ? TRUE : FALSE; + 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 */ /******************************************************************************/ -/* 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 = + (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, 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; 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. - If it is not found then create it and push it into the rendez-vous point */ - comm = SIMIX_rdv_get_request(rdv, filter, arg); - - if(comm == NULL){ - comm = SIMIX_communication_new(comm_recv, rdv); + /*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) { + 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; 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 */ - /* 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) +/** @Returns whether the (asynchronous) communication is done yet or not */ +XBT_INLINE int SIMIX_network_test(smx_comm_t comm) { - if (timeout > 0) - SIMIX_cond_wait_timeout(rvp_cond, rvp_comm_mutex, timeout - start_time); - else - SIMIX_cond_wait(rvp_cond, rvp_comm_mutex); - + /*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) { - if(SIMIX_action_get_state (comm) == SURF_ACTION_DONE){ - memcpy(comm->data - - return SIMIX_action_get_state (comm) == SURF_ACTION_DONE ? TRUE : FALSE; -}*/ - - - - - - - + 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; +}