#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)");
* \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));
return comm;
}
- /* no relevant request found. Return NULL */
DEBUG0("Communication request not found");
return NULL;
}
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);
}
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 */
*/
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_sem_acquire_timeout(comm->sem, 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 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);
- }
- RETHROW;
- }
+ 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);
}
+ /* 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_communication_destroy(comm);
THROW0(network_error, 0, "Link failure");
}
+ SIMIX_communication_destroy(comm);
}
/**
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 Copy the communication data from the sender's buffer to the receiver's one
* \param comm The communication
if(!comm->src_buff || !comm->dst_buff)
return;
- size_t src_buff_size = comm->src_buff_size;
- size_t dst_buff_size = *comm->dst_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);
+ 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 = dst_buff_size;
+ *comm->dst_buff_size = buff_size;
- if(dst_buff_size == 0)
+ if(buff_size == 0)
return;
-
- 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);
+ 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,"%zu %zu\n",key,data);
+ }
+ fclose(out);
}
/**
XBT_INLINE void SIMIX_network_wait(smx_comm_t comm, double timeout) {
/* Wait for communication completion */
SIMIX_communication_wait_for_completion(comm, timeout);
-
- SIMIX_communication_destroy(comm);
}
/** @Returns whether the (asynchronous) communication is done yet or not */
SIMIX_communication_destroy(comm_finished);
THROW0(network_error, 0, "Link failure");
}
+ SIMIX_communication_destroy(comm_finished);
return found_comm;
}
