#include "private.h"
#include "xbt/sysdep.h"
#include "xbt/log.h"
-XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gos, msg,
+XBT_LOG_NEW_DEFAULT_SUBCATEGORY(msg_gos, msg,
"Logging specific to MSG (gos)");
/** \defgroup msg_gos_functions MSG Operating System Functions
if(max_duration>0) {
if(!first_time) {
- MSG_RETURN(MSG_OK);
+ MSG_RETURN(MSG_TRANSFER_FAILURE);
}
}
- xbt_assert2(!(h_simdata->sleeping[channel]),
- "A process (%s(%d)) is already blocked on this channel",
+ xbt_assert3(!(h_simdata->sleeping[channel]),
+ "A process (%s(%d)) is already blocked on channel %d",
h_simdata->sleeping[channel]->name,
- h_simdata->sleeping[channel]->simdata->PID);
+ h_simdata->sleeping[channel]->simdata->PID,
+ channel);
h_simdata->sleeping[channel] = process; /* I'm waiting. Wake me up when you're ready */
if(max_duration>0) {
- __MSG_process_block(max_duration);
+ __MSG_process_block(max_duration,"");
} else {
- __MSG_process_block(-1);
+ __MSG_process_block(-1,"");
}
h_simdata->sleeping[channel] = NULL;
first_time = 0;
__MSG_process_unblock(t_simdata->sender);
}
- PAJE_PROCESS_PUSH_STATE(process,"C");
+ PAJE_PROCESS_PUSH_STATE(process,"C",t);
do {
DEBUG0("Waiting for action termination");
listening. This value has to be >=0 and < than the maximal
number of channels fixed with MSG_set_channel_number().
* \param max_duration the maximum time to wait for a task before giving
- up. In such a case, \a task will not be modified and will still be
- equal to \c NULL when returning.
+ up. In such a case, #MSG_TRANSFER_FAILURE will be returned, \a task
+ will not be modified and will still be
+ equal to \c NULL when returning.
* \return #MSG_FATAL if \a task is equal to \c NULL, #MSG_WARNING
if \a *task is not equal to \c NULL, and #MSG_OK otherwise.
*/
h_simdata->sleeping[channel]->simdata->PID);
h_simdata->sleeping[channel] = process; /* I'm waiting. Wake me up when you're ready */
if(max_duration>0) {
- __MSG_process_block(max_duration);
+ __MSG_process_block(max_duration,"");
} else {
- __MSG_process_block(-1);
+ __MSG_process_block(-1,"");
}
if(surf_workstation_resource->extension_public->get_state(h_simdata->host)
== SURF_CPU_OFF) {
return count;
}
-/** \ingroup msg_gos_functions
- * \brief Put a task on a channel of an host and waits for the end of the
- * transmission.
+/** \ingroup msg_gos_functions \brief Put a task on a channel of an
+ * host (with a timeout on the waiting of the destination host) and
+ * waits for the end of the transmission.
*
* This function is used for describing the behavior of an agent. It
- * takes three parameter.
+ * takes four parameter.
* \param task a #m_task_t to send on another location. This task
will not be usable anymore when the function will return. There is
no automatic task duplication and you have to save your parameters
* \param channel the channel on which the agent should put this
task. This value has to be >=0 and < than the maximal number of
channels fixed with MSG_set_channel_number().
+ * \param max_duration the maximum time to wait for a task before giving
+ up. In such a case, #MSG_TRANSFER_FAILURE will be returned, \a task
+ will not be modified
* \return #MSG_FATAL if \a task is not properly initialized and
- * #MSG_OK otherwise.
+ #MSG_OK otherwise. Returns #MSG_HOST_FAILURE if the host on which
+ this function was called was shut down. Returns
+ #MSG_TRANSFER_FAILURE if the transfer could not be properly done
+ (network failure, dest failure, timeout...)
*/
-MSG_error_t MSG_task_put(m_task_t task,
- m_host_t dest, m_channel_t channel)
+MSG_error_t MSG_task_put_with_timeout(m_task_t task, m_host_t dest,
+ m_channel_t channel, double max_duration)
{
m_process_t process = MSG_process_self();
simdata_task_t task_simdata = NULL;
e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
m_host_t local_host = NULL;
m_host_t remote_host = NULL;
+ int first_time = 1;
CHECK_HOST();
process->simdata->put_host = dest;
process->simdata->put_channel = channel;
while(!(task_simdata->comm)) {
+ if(max_duration>0) {
+ if(!first_time) {
+ MSG_RETURN(MSG_TRANSFER_FAILURE);
+ }
+ }
DEBUG0("Communication not initiated yet. Let's block!");
- __MSG_process_block(-1);
+ if(max_duration>0)
+ __MSG_process_block(max_duration,task->name);
+ else
+ __MSG_process_block(-1,task->name);
+
+ first_time = 0;
+
+ if(surf_workstation_resource->extension_public->
+ get_state(local_host->simdata->host) == SURF_CPU_OFF) {
+ xbt_fifo_remove(((simdata_host_t) remote_host->simdata)->mbox[channel],
+ task);
+ MSG_task_destroy(task);
+ MSG_RETURN(MSG_HOST_FAILURE);
+ }
}
DEBUG0("Registering to this communication");
surf_workstation_resource->common_public->action_use(task_simdata->comm);
process->simdata->put_channel = -1;
- PAJE_PROCESS_PUSH_STATE(process,"C");
+ PAJE_PROCESS_PUSH_STATE(process,"C",task);
state=surf_workstation_resource->common_public->action_get_state(task_simdata->comm);
while (state==SURF_ACTION_RUNNING) {
MSG_RETURN(MSG_TRANSFER_FAILURE);
}
}
+/** \ingroup msg_gos_functions
+ * \brief Put a task on a channel of an host and waits for the end of the
+ * transmission.
+ *
+ * This function is used for describing the behavior of an agent. It
+ * takes three parameter.
+ * \param task a #m_task_t to send on another location. This task
+ will not be usable anymore when the function will return. There is
+ no automatic task duplication and you have to save your parameters
+ before calling this function. Tasks are unique and once it has been
+ sent to another location, you should not access it anymore. You do
+ not need to call MSG_task_destroy() but to avoid using, as an
+ effect of inattention, this task anymore, you definitely should
+ renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
+ can be transfered iff it has been correctly created with
+ MSG_task_create().
+ * \param dest the destination of the message
+ * \param channel the channel on which the agent should put this
+ task. This value has to be >=0 and < than the maximal number of
+ channels fixed with MSG_set_channel_number().
+ * \return #MSG_FATAL if \a task is not properly initialized and
+ * #MSG_OK otherwise. Returns #MSG_HOST_FAILURE if the host on which
+ * this function was called was shut down. Returns
+ * #MSG_TRANSFER_FAILURE if the transfer could not be properly done
+ * (network failure, dest failure)
+ */
+MSG_error_t MSG_task_put(m_task_t task,
+ m_host_t dest, m_channel_t channel)
+{
+ return MSG_task_put_with_timeout(task, dest, channel, -1.0);
+}
/** \ingroup msg_gos_functions
* \brief Does exactly the same as MSG_task_put but with a bounded transmition
__MSG_task_execute(process, task);
- PAJE_PROCESS_PUSH_STATE(process,"E");
+ PAJE_PROCESS_PUSH_STATE(process,"E",task);
res = __MSG_wait_for_computation(process,task);
PAJE_PROCESS_POP_STATE(process);
return res;
CHECK_HOST();
simdata = task->simdata;
- xbt_assert0(!simdata->compute,"This taks is executed somewhere else. Go fix your code!");
-
+ xbt_assert0((!simdata->compute)&&(task->simdata->using==1),
+ "This taks is executed somewhere else. Go fix your code!");
+ task->simdata->using++;
simdata->compute = surf_workstation_resource->extension_public->
execute(MSG_process_get_host(process)->simdata->host,
simdata->computation_amount);
set_priority(simdata->compute, simdata->priority);
surf_workstation_resource->common_public->action_set_data(simdata->compute,task);
+ task->simdata->using--;
}
MSG_error_t __MSG_wait_for_computation(m_process_t process, m_task_t task)
