Logo AND Algorithmique Numérique Distribuée

Public GIT Repository
Bounded receive
[simgrid.git] / src / msg / msg_gos.c
index 2ceacc2..7eb2d79 100644 (file)
@@ -75,7 +75,7 @@ msg_error_t MSG_parallel_task_execute(msg_task_t task)
       XBT_DEBUG("Parallel execution action created: %p", simdata->compute);
     } else {
       simdata->compute = simcall_host_execute(task->name,
-                                              p_simdata->m_host->smx_host,
+                                              p_simdata->m_host,
                                               simdata->computation_amount,
                                               simdata->priority);
 
@@ -164,7 +164,7 @@ msg_error_t MSG_process_sleep(double nb_sec)
  *    sometime waiting on all these mailboxes using @ref MSG_comm_waitany. You can find
  *    an example of use of this function in the @ref MSG_examples section.
  *  - Provide a proper patch to implement this functionality back in MSG. That wouldn't be
- *    very difficult actually. Check the function #MSG_mailbox_get_task_ext. During its call to
+ *    very difficult actually. Check the function @ref MSG_mailbox_get_task_ext. During its call to
  *    simcall_comm_recv(), the 5th argument, match_fun, is NULL. Create a function that filters
  *    messages according to the host (that you will pass as sixth argument to simcall_comm_recv()
  *    and that your filtering function will receive as first parameter, and then, the filter could
@@ -186,6 +186,26 @@ MSG_task_receive_from_host(msg_task_t * task, const char *alias,
   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.
  *
@@ -205,6 +225,22 @@ msg_error_t MSG_task_receive(msg_task_t * task, const char *alias)
   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.
  *
@@ -228,6 +264,25 @@ MSG_task_receive_with_timeout(msg_task_t * task, const char *alias,
   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.
  *
@@ -256,6 +311,31 @@ MSG_task_receive_ext(msg_task_t * task, const char *alias, double timeout,
                                   host, timeout);
 }
 
+/** \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);
+}
+
 /** \ingroup msg_task_usage
  * \brief Sends a task on a mailbox.
  *
@@ -271,6 +351,25 @@ msg_comm_t MSG_task_isend(msg_task_t task, const char *alias)
   return MSG_task_isend_with_matching(task,alias,NULL,NULL);
 }
 
+/** \ingroup msg_task_usage
+ * \brief Sends a task on a mailbox with a maximum rate
+ *
+ * This is a non blocking function: use MSG_comm_wait() or MSG_comm_test()
+ * to end the communication. The maxrate parameter allows the application
+ * to limit the bandwidth utilization of network links when sending the task.
+ *
+ * \param task a #msg_task_t to send on another location.
+ * \param alias name of the mailbox to sent the task to
+ * \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)
+{
+  task->simdata->rate = maxrate;
+  return MSG_task_isend_with_matching(task,alias,NULL,NULL);
+}
+
+
 /** \ingroup msg_task_usage
  * \brief Sends a task on a mailbox, with support for matching requests
  *
@@ -294,7 +393,9 @@ XBT_INLINE msg_comm_t MSG_task_isend_with_matching(msg_task_t task, const char *
   msg_process_t process = MSG_process_self();
   msg_mailbox_t mailbox = MSG_mailbox_get_by_alias(alias);
 
-  /* FIXME: these functions are not traceable */
+#ifdef HAVE_TRACING
+  int call_end = TRACE_msg_task_put_start(task);
+#endif
 
   /* Prepare the task to send */
   t_simdata = task->simdata;
@@ -317,6 +418,16 @@ XBT_INLINE msg_comm_t MSG_task_isend_with_matching(msg_task_t task, const char *
     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;
 }
@@ -376,6 +487,66 @@ void MSG_task_dsend(msg_task_t task, const char *alias, void_f_pvoid_t cleanup)
 #endif
 }
 
+
+/** \ingroup msg_task_usage
+ * \brief Sends a task on a mailbox with a maximal rate.
+ *
+ * This is a non blocking detached send function.
+ * Think of it as a best effort send. Keep in mind that the third parameter
+ * is only called if the communication fails. If the communication does work,
+ * it is responsibility of the receiver code to free anything related to
+ * the task, as usual. More details on this can be obtained on
+ * <a href="http://lists.gforge.inria.fr/pipermail/simgrid-user/2011-November/002649.html">this thread</a>
+ * in the SimGrid-user mailing list archive.
+ *
+ * \param task a #msg_task_t to send on another location.
+ * \param alias name of the mailbox to sent the task to
+ * \param cleanup a function to destroy the task if the
+ * 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)
+{
+  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
+}
+
 /** \ingroup msg_task_usage
  * \brief Starts listening for receiving a task from an asynchronous communication.
  *
@@ -409,6 +580,40 @@ msg_comm_t MSG_task_irecv(msg_task_t *task, const char *name)
   return comm;
 }
 
+/** \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);
+  simcall_comm_change_rate_first_action(rdv,rate);
+  /* FIXME: these functions are not traceable */
+
+  /* Sanity check */
+  xbt_assert(task, "Null pointer for the task storage");
+
+  if (*task)
+    XBT_CRITICAL
+        ("MSG_task_irecv() was asked to write in a non empty task struct.");
+
+  /* Try to receive it by calling SIMIX network layer */
+  msg_comm_t comm = xbt_new0(s_msg_comm_t, 1);
+  comm->task_sent = NULL;
+  comm->task_received = task;
+  comm->status = MSG_OK;
+  comm->s_comm = simcall_comm_irecv(rdv, task, NULL, NULL, NULL);
+
+  return comm;
+}
+
 /** \ingroup msg_task_usage
  * \brief Checks whether a communication is done, and if yes, finalizes it.
  * \param comm the communication to test
@@ -421,6 +626,7 @@ int MSG_comm_test(msg_comm_t comm)
 {
   xbt_ex_t e;
   int finished = 0;
+
   TRY {
     finished = simcall_comm_test(comm->s_comm);
 
@@ -652,7 +858,7 @@ msg_task_t MSG_comm_get_task(msg_comm_t comm)
 }
 
 /**
- * \brief This function is called by SIMIX to copy the data of a comm.
+ * \brief This function is called by SIMIX in kernel mode to copy the data of a comm.
  * \param comm the comm
  * \param buff the data copied
  * \param buff_size size of the buffer
@@ -731,6 +937,29 @@ MSG_task_send_with_timeout(msg_task_t task, const char *alias,
                                       task, timeout);
 }
 
+/** \ingroup msg_task_usage
+ * \brief Sends a task to a mailbox with a timeout and with a maximum rate
+ *
+ * This is a blocking function, the execution flow will be blocked
+ * until the task is sent or the timeout is achieved.
+ *
+ * \param task the task to be sent
+ * \param alias the mailbox name to where the task is sent
+ * \param timeout is the maximum wait time for completion (if -1, this call is the same as #MSG_task_send)
+ * \param maxrate the maximum communication rate for sending this task
+ *
+ * \return Returns #MSG_OK if the task was successfully sent,
+ * #MSG_HOST_FAILURE, or #MSG_TRANSFER_FAILURE, or #MSG_TIMEOUT otherwise.
+ */
+msg_error_t
+MSG_task_send_with_timeout_bounded(msg_task_t task, const char *alias,
+                           double timeout, double maxrate)
+{
+  task->simdata->rate = maxrate;
+  return MSG_mailbox_put_with_timeout(MSG_mailbox_get_by_alias(alias),
+                                      task, timeout);
+}
+
 /** \ingroup msg_task_usage
  * \brief Check if there is a communication going on in a mailbox.
  *