-/* Copyright (c) 2004-2012. The SimGrid Team. All rights reserved. */
+/* Copyright (c) 2004-2014. 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. */
*/
msg_error_t MSG_task_execute(msg_task_t task)
{
- return MSG_parallel_task_execute(task);
+ /* TODO: add this to other locations */
+ msg_host_t host = MSG_process_get_host(MSG_process_self());
+ MSG_host_add_task(host, task);
+
+ msg_error_t ret = MSG_parallel_task_execute(task);
+
+ MSG_host_del_task(host, task);
+
+ return ret;
}
/** \ingroup msg_task_usage
xbt_assert((!simdata->compute) && (task->simdata->isused == 0),
"This task is executed somewhere else. Go fix your code! %d",
- task->simdata->isused);
+ task->simdata->isused!=NULL);
XBT_DEBUG("Computing on %s", MSG_process_get_name(MSG_process_self()));
TRY {
-
- simdata->isused=1;
+ if (msg_global->multiple_backtraces)
+ MSG_BT(simdata->isused, "Using Backtrace");
+ else
+ simdata->isused = (void*)1;
if (simdata->host_nb > 0) {
simdata->compute = simcall_host_parallel_execute(task->name,
1.0, -1.0);
XBT_DEBUG("Parallel execution action created: %p", simdata->compute);
} else {
+ unsigned long affinity_mask = (unsigned long) xbt_dict_get_or_null_ext(simdata->affinity_mask_db, (char *) p_simdata->m_host, sizeof(msg_host_t));
+ XBT_DEBUG("execute %s@%s with affinity(0x%04lx)", MSG_task_get_name(task), MSG_host_get_name(p_simdata->m_host), affinity_mask);
+
simdata->compute = simcall_host_execute(task->name,
p_simdata->m_host,
simdata->computation_amount,
- simdata->priority);
+ simdata->priority,
+ simdata->bound,
+ affinity_mask
+ );
}
#ifdef HAVE_TRACING
case cancel_error:
status = MSG_TASK_CANCELED;
break;
+ case host_error:
+ status = MSG_HOST_FAILURE;
+ break;
default:
RETHROW;
}
*/
msg_error_t MSG_process_sleep(double nb_sec)
{
+ xbt_ex_t e;
msg_error_t status = MSG_OK;
/*msg_process_t proc = MSG_process_self();*/
/*proc->simdata->waiting_action = act_sleep;
FIXME: check if not setting the waiting_action breaks something on msg
-
+
proc->simdata->waiting_action = NULL;*/
- simcall_process_sleep(nb_sec);
+ TRY {
+ simcall_process_sleep(nb_sec);
+ }
+ CATCH(e) {
+ switch (e.category) {
+ case cancel_error:
+ status = MSG_TASK_CANCELED;
+ break;
+ default:
+ RETHROW;
+ }
+ xbt_ex_free(e);
+ }
#ifdef HAVE_TRACING
TRACE_msg_process_sleep_out(MSG_process_self());
return MSG_task_receive_ext(task, alias, -1, host);
}
+/** msg_task_usage
+ *\brief Deprecated function that used to receive a task from a mailbox from a specific host
+ *\brief at a given rate
+ *
+ * \param task a memory location for storing a #msg_task_t.
+ * \param alias name of the mailbox to receive the task from
+ * \param host a #msg_host_t host from where the task was sent
+ * \param rate limit the reception to rate bandwidth
+ *
+ * \return Returns
+ * #MSG_OK if the task was successfully received,
+ * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
+ */
+msg_error_t
+MSG_task_receive_from_host_bounded(msg_task_t * task, const char *alias,
+ msg_host_t host, double rate)
+{
+ return MSG_task_receive_ext_bounded(task, alias, -1, host, rate);
+}
+
/** \ingroup msg_task_usage
* \brief Receives a task from a mailbox.
*
return MSG_task_receive_with_timeout(task, alias, -1);
}
+/** \ingroup msg_task_usage
+ * \brief Receives a task from a mailbox at a given rate.
+ *
+ * \param task a memory location for storing a #msg_task_t.
+ * \param alias name of the mailbox to receive the task from
+ * \param rate limit the reception to rate bandwidth
+ *
+ * \return Returns
+ * #MSG_OK if the task was successfully received,
+ * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE otherwise.
+ */
+msg_error_t MSG_task_receive_bounded(msg_task_t * task, const char *alias,
+ double rate)
+{
+ return MSG_task_receive_with_timeout_bounded(task, alias, -1, rate);
+}
+
/** \ingroup msg_task_usage
* \brief Receives a task from a mailbox with a given timeout.
*
return MSG_task_receive_ext(task, alias, timeout, NULL);
}
+/** \ingroup msg_task_usage
+ * \brief Receives a task from a mailbox with a given timeout and at a given rate.
+ *
+ * \param task a memory location for storing a #msg_task_t.
+ * \param alias name of the mailbox to receive the task from
+ * \param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_receive)
+ * \param rate limit the reception to rate bandwidth
+ *
+ * \return Returns
+ * #MSG_OK if the task was successfully received,
+ * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
+ */
+msg_error_t
+MSG_task_receive_with_timeout_bounded(msg_task_t * task, const char *alias,
+ double timeout,double rate)
+{
+ return MSG_task_receive_ext_bounded(task, alias, timeout, NULL, rate);
+}
+
/** \ingroup msg_task_usage
* \brief Receives a task from a mailbox from a specific host with a given timeout.
*
MSG_task_receive_ext(msg_task_t * task, const char *alias, double timeout,
msg_host_t host)
{
+ xbt_ex_t e;
+ msg_error_t ret = MSG_OK;
XBT_DEBUG
("MSG_task_receive_ext: Trying to receive a message on mailbox '%s'",
alias);
- return MSG_mailbox_get_task_ext(MSG_mailbox_get_by_alias(alias), task,
- host, timeout);
+ TRY {
+ ret = MSG_mailbox_get_task_ext(MSG_mailbox_get_by_alias(alias), task,
+ host, timeout);
+ }
+ CATCH(e) {
+ switch (e.category) {
+ case cancel_error: /* may be thrown by MSG_mailbox_get_by_alias */
+ ret = MSG_HOST_FAILURE;
+ break;
+ default:
+ RETHROW;
+ }
+ xbt_ex_free(e);
+ }
+ return ret;
+}
+
+/** \ingroup msg_task_usage
+ * \brief Receives a task from a mailbox from a specific host with a given timeout
+ * and at a given rate.
+ *
+ * \param task a memory location for storing a #msg_task_t.
+ * \param alias name of the mailbox to receive the task from
+ * \param timeout is the maximum wait time for completion (provide -1 for no timeout)
+ * \param host a #msg_host_t host from where the task was sent
+ * \param rate limit the reception to rate bandwidth
+ *
+ * \return Returns
+ * #MSG_OK if the task was successfully received,
+* #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
+ */
+msg_error_t
+MSG_task_receive_ext_bounded(msg_task_t * task, const char *alias, double timeout,
+ msg_host_t host, double rate)
+{
+ XBT_DEBUG
+ ("MSG_task_receive_ext: Trying to receive a message on mailbox '%s'",
+ alias);
+ return MSG_mailbox_get_task_ext_bounded(MSG_mailbox_get_by_alias(alias), task,
+ host, timeout, rate);
+}
+
+/* Internal function used to factorize code between
+ * MSG_task_isend_with_matching() and MSG_task_dsend().
+ */
+static XBT_INLINE
+msg_comm_t MSG_task_isend_internal(msg_task_t task, const char *alias,
+ int (*match_fun)(void*,void*, smx_action_t),
+ void *match_data, void_f_pvoid_t cleanup,
+ int detached)
+{
+ simdata_task_t t_simdata = NULL;
+ msg_process_t process = MSG_process_self();
+ msg_mailbox_t mailbox = MSG_mailbox_get_by_alias(alias);
+
+#ifdef HAVE_TRACING
+ int call_end = TRACE_msg_task_put_start(task);
+#endif
+
+ /* Prepare the task to send */
+ t_simdata = task->simdata;
+ t_simdata->sender = process;
+ t_simdata->source = ((simdata_process_t) SIMIX_process_self_get_data(process))->m_host;
+
+ if (t_simdata->isused != 0) {
+ if (msg_global->multiple_backtraces){
+ XBT_ERROR("This task is already used in there:");
+ xbt_backtrace_display(t_simdata->isused);
+ THROWF(unknown_error, 0, "And you try to reuse it from here. You cannot send it now. Go fix your code!");
+ } else {
+ THROWF(unknown_error, 0, "This task is still being used somewhere else. You cannot send it now. Go fix your code! (use --cfg=msg/multiple_backtraces:on to get the backtrace of the other process)");
+ }
+ }
+
+ if (msg_global->multiple_backtraces)
+ MSG_BT(t_simdata->isused, "Using Backtrace");
+ else
+ t_simdata->isused = (void*)1;
+ t_simdata->comm = NULL;
+ msg_global->sent_msg++;
+
+ /* Send it by calling SIMIX network layer */
+ smx_action_t act = simcall_comm_isend(mailbox, t_simdata->message_size,
+ t_simdata->rate, task, sizeof(void *),
+ match_fun, cleanup, match_data,detached);
+ t_simdata->comm = act; /* FIXME: is the field t_simdata->comm still useful? */
+
+ msg_comm_t comm;
+ if (detached) {
+ comm = NULL;
+ } else {
+ comm = xbt_new0(s_msg_comm_t, 1);
+ comm->task_sent = task;
+ comm->task_received = NULL;
+ comm->status = MSG_OK;
+ comm->s_comm = act;
+ }
+
+#ifdef HAVE_TRACING
+ if (TRACE_is_enabled()) {
+ simcall_set_category(act, task->category);
+ }
+
+ if (call_end)
+ TRACE_msg_task_put_end();
+#endif
+
+ return comm;
}
+
/** \ingroup msg_task_usage
* \brief Sends a task on a mailbox.
*
*/
msg_comm_t MSG_task_isend(msg_task_t task, const char *alias)
{
- return MSG_task_isend_with_matching(task,alias,NULL,NULL);
+ return MSG_task_isend_internal(task, alias, NULL, NULL, NULL, 0);
}
/** \ingroup msg_task_usage
* \param maxrate the maximum communication rate for sending this task .
* \return the msg_comm_t communication created
*/
-msg_comm_t MSG_task_isend_bounded(msg_task_t task, const char *alias, double maxrate)
+msg_comm_t MSG_task_isend_bounded(msg_task_t task, const char *alias,
+ double maxrate)
{
task->simdata->rate = maxrate;
- return MSG_task_isend_with_matching(task,alias,NULL,NULL);
+ return MSG_task_isend_internal(task, alias, NULL, NULL, NULL, 0);
}
* \param match_data user provided data passed to match_fun
* \return the msg_comm_t communication created
*/
-XBT_INLINE msg_comm_t MSG_task_isend_with_matching(msg_task_t task, const char *alias,
- int (*match_fun)(void*,void*, smx_action_t),
- void *match_data)
+msg_comm_t MSG_task_isend_with_matching(msg_task_t task, const char *alias,
+ int (*match_fun)(void*, void*,
+ smx_action_t),
+ void *match_data)
{
- simdata_task_t t_simdata = NULL;
- msg_process_t process = MSG_process_self();
- msg_mailbox_t mailbox = MSG_mailbox_get_by_alias(alias);
-
-#ifdef HAVE_TRACING
- int call_end = TRACE_msg_task_put_start(task);
-#endif
-
- /* Prepare the task to send */
- t_simdata = task->simdata;
- t_simdata->sender = process;
- t_simdata->source = ((simdata_process_t) SIMIX_process_self_get_data(process))->m_host;
-
- xbt_assert(t_simdata->isused == 0,
- "This task is still being used somewhere else. You cannot send it now. Go fix your code!");
-
- t_simdata->isused = 1;
- t_simdata->comm = NULL;
- msg_global->sent_msg++;
-
- /* Send it by calling SIMIX network layer */
- msg_comm_t comm = xbt_new0(s_msg_comm_t, 1);
- comm->task_sent = task;
- comm->task_received = NULL;
- comm->status = MSG_OK;
- comm->s_comm =
- simcall_comm_isend(mailbox, t_simdata->message_size,
- t_simdata->rate, task, sizeof(void *), match_fun, NULL, match_data, 0);
- t_simdata->comm = comm->s_comm; /* FIXME: is the field t_simdata->comm still useful? */
-#ifdef HAVE_TRACING
- if (TRACE_is_enabled()) {
- simcall_set_category(comm->s_comm, task->category);
- }
-#endif
-
-#ifdef HAVE_TRACING
- if (call_end)
- TRACE_msg_task_put_end();
-#endif
-
- return comm;
+ return MSG_task_isend_internal(task, alias, match_fun, match_data, NULL, 0);
}
/** \ingroup msg_task_usage
*/
void MSG_task_dsend(msg_task_t task, const char *alias, void_f_pvoid_t cleanup)
{
- simdata_task_t t_simdata = NULL;
- msg_process_t process = MSG_process_self();
- msg_mailbox_t mailbox = MSG_mailbox_get_by_alias(alias);
-
- /* Prepare the task to send */
- t_simdata = task->simdata;
- t_simdata->sender = process;
- t_simdata->source = ((simdata_process_t) SIMIX_process_self_get_data(process))->m_host;
-
- xbt_assert(t_simdata->isused == 0,
- "This task is still being used somewhere else. You cannot send it now. Go fix your code!");
-
- t_simdata->isused = 1;
- t_simdata->comm = NULL;
- msg_global->sent_msg++;
-
-#ifdef HAVE_TRACING
- int call_end = TRACE_msg_task_put_start(task);
-#endif
-
- /* Send it by calling SIMIX network layer */
- smx_action_t comm = simcall_comm_isend(mailbox, t_simdata->message_size,
- t_simdata->rate, task, sizeof(void *), NULL, cleanup, NULL, 1);
- t_simdata->comm = comm;
-#ifdef HAVE_TRACING
- if (TRACE_is_enabled()) {
- simcall_set_category(comm, task->category);
- }
-#endif
-
-#ifdef HAVE_TRACING
- if (call_end)
- TRACE_msg_task_put_end();
-#endif
+ MSG_task_isend_internal(task, alias, NULL, NULL, cleanup, 1);
}
-
/** \ingroup msg_task_usage
* \brief Sends a task on a mailbox with a maximal rate.
*
* communication fails, e.g. MSG_task_destroy
* (if NULL, no function will be called)
* \param maxrate the maximum communication rate for sending this task
- *
+ *
*/
-void MSG_task_dsend_bounded(msg_task_t task, const char *alias, void_f_pvoid_t cleanup, double maxrate)
+void MSG_task_dsend_bounded(msg_task_t task, const char *alias,
+ void_f_pvoid_t cleanup, double maxrate)
{
task->simdata->rate = maxrate;
-
- simdata_task_t t_simdata = NULL;
- msg_process_t process = MSG_process_self();
- msg_mailbox_t mailbox = MSG_mailbox_get_by_alias(alias);
-
- /* Prepare the task to send */
- t_simdata = task->simdata;
- t_simdata->sender = process;
- t_simdata->source = ((simdata_process_t) SIMIX_process_self_get_data(process))->m_host;
-
- xbt_assert(t_simdata->isused == 0,
- "This task is still being used somewhere else. You cannot send it now. Go fix your code!");
-
- t_simdata->isused = 1;
- t_simdata->comm = NULL;
- msg_global->sent_msg++;
-
-#ifdef HAVE_TRACING
- int call_end = TRACE_msg_task_put_start(task);
-#endif
-
- /* Send it by calling SIMIX network layer */
- smx_action_t comm = simcall_comm_isend(mailbox, t_simdata->message_size,
- t_simdata->rate, task, sizeof(void *), NULL, cleanup, NULL, 1);
- t_simdata->comm = comm;
-#ifdef HAVE_TRACING
- if (TRACE_is_enabled()) {
- simcall_set_category(comm, task->category);
- }
-#endif
-
-#ifdef HAVE_TRACING
- if (call_end)
- TRACE_msg_task_put_end();
-#endif
+ MSG_task_dsend(task, alias, cleanup);
}
/** \ingroup msg_task_usage
*
* This is a non blocking function: use MSG_comm_wait() or MSG_comm_test()
* to end the communication.
- *
+ *
* \param task a memory location for storing a #msg_task_t. has to be valid until the end of the communication.
* \param name of the mailbox to receive the task on
* \return the msg_comm_t communication created
*/
msg_comm_t MSG_task_irecv(msg_task_t *task, const char *name)
+{
+ return MSG_task_irecv_bounded(task, name, -1.0);
+}
+
+/** \ingroup msg_task_usage
+ * \brief Starts listening for receiving a task from an asynchronous communication
+ * at a given rate.
+ *
+ * \param task a memory location for storing a #msg_task_t. has to be valid until the end of the communication.
+ * \param name of the mailbox to receive the task on
+ * \param rate limit the bandwidth to the given rate
+ * \return the msg_comm_t communication created
+ */
+msg_comm_t MSG_task_irecv_bounded(msg_task_t *task, const char *name,
+ double rate)
{
smx_rdv_t rdv = MSG_mailbox_get_by_alias(name);
comm->task_sent = NULL;
comm->task_received = task;
comm->status = MSG_OK;
- comm->s_comm = simcall_comm_irecv(rdv, task, NULL, NULL, NULL);
+ comm->s_comm = simcall_comm_irecv(rdv, task, NULL, NULL, NULL, rate);
return comm;
}
*
* It takes two parameters.
* \param comm the communication to wait.
- * \param timeout Wait until the communication terminates or the timeout
+ * \param timeout Wait until the communication terminates or the timeout
* occurs. You can provide a -1 timeout to obtain an infinite timeout.
* \return msg_error_t
*/
&& (channel < msg_global->max_channel), "Invalid channel %d",
channel);
- XBT_DEBUG("MSG_task_put_with_timout: Trying to send a task to '%s'", SIMIX_host_get_name(dest->smx_host));
+ XBT_DEBUG("MSG_task_put_with_timout: Trying to send a task to '%s'", MSG_host_get_name(dest));
return
MSG_mailbox_put_with_timeout(MSG_mailbox_get_by_channel
(dest, channel), task, timeout);
