src/msg/gos.c

   1 /*      $Id$     */
   2
   3 /* Copyright (c) 2002,2003,2004 Arnaud Legrand. All rights reserved.        */
   4
   5 /* This program is free software; you can redistribute it and/or modify it
   6  * under the terms of the license (GNU LGPL) which comes with this package. */
   7
   8 #include"private.h"
   9 #include"xbt/sysdep.h"
  10 #include "xbt/error.h"
  11 XBT_LOG_NEW_DEFAULT_SUBCATEGORY(gos, msg,
  12                                 "Logging specific to MSG (gos)");
  13
  14 /** \ingroup msg_gos_functions
  15  * \brief This function is now deprecated and useless. Please stop using it.
  16  */
  17
  18 MSG_error_t MSG_process_start(m_process_t process)
  19 {
  20   xbt_assert0(0,"This function is now deprecated and useless. Please stop using it.");
  21
  22   return MSG_OK;
  23 }
  24
  25 /** \ingroup msg_gos_functions
  26  * \brief Listen on a channel and wait for receiving a task.
  27  *
  28  * It takes two parameter.
  29  * \param task a memory location for storing a #m_task_t. It will
  30    hold a task when this function will return. Thus \a task should not
  31    be equal to \c NULL and \a *task should be equal to \c NULL. If one of
  32    those two condition does not hold, there will be a warning message.
  33  * \param channel the channel on which the agent should be
  34    listening. This value has to be >=0 and < than the maximal
  35    number of channels fixed with MSG_set_channel_number().
  36  * \return #MSG_FATAL if \a task is equal to \c NULL, #MSG_WARNING
  37  * if \a *task is not equal to \c NULL, and #MSG_OK otherwise.
  38  */
  39 MSG_error_t MSG_task_get(m_task_t * task,
  40                          m_channel_t channel)
  41 {
  42   m_process_t process = MSG_process_self();
  43   m_task_t t = NULL;
  44   m_host_t h = NULL;
  45   simdata_task_t t_simdata = NULL;
  46   simdata_host_t h_simdata = NULL;
  47   int warning = 0;
  48   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
  49
  50   CHECK_HOST();
  51   /* Sanity check */
  52   xbt_assert0(task,"Null pointer for the task\n");
  53
  54   if (*task)
  55     CRITICAL0("MSG_task_get() was asked to write in a non empty task struct.");
  56
  57   /* Get the task */
  58   h = MSG_host_self();
  59   h_simdata = h->simdata;
  60   while ((t = xbt_fifo_pop(h_simdata->mbox[channel])) == NULL) {
  61     xbt_assert0(!(h_simdata->sleeping[channel]),
  62                 "A process is already blocked on this channel");
  63     h_simdata->sleeping[channel] = process; /* I'm waiting. Wake me up when you're ready */
  64     MSG_process_suspend(process);
  65     if(surf_workstation_resource->extension_public->get_state(h_simdata->host)
  66        == SURF_CPU_OFF)
  67       MSG_RETURN(MSG_HOST_FAILURE);
  68     h_simdata->sleeping[channel] = NULL;
  69     /* OK, we should both be ready now. Are you there ? */
  70   }
  71
  72   t_simdata = t->simdata;
  73   /*   *task = __MSG_task_copy(t); */
  74   *task=t;
  75
  76   /* Transfer */
  77   t_simdata->using++;
  78   t_simdata->comm = surf_workstation_resource->extension_public->
  79     communicate(MSG_process_get_host(t_simdata->sender)->simdata->host,
  80                 h, t_simdata->message_size);
  81
  82   do {
  83     __MSG_task_wait_event(process, t);
  84     state=surf_workstation_resource->common_public->action_get_state(t_simdata->comm);
  85   } while (state==SURF_ACTION_RUNNING);
  86
  87   if(state == SURF_ACTION_DONE) MSG_RETURN(MSG_OK);
  88   else if(surf_workstation_resource->extension_public->get_state(h_simdata->host)
  89           == SURF_CPU_OFF)
  90     MSG_RETURN(MSG_HOST_FAILURE);
  91   else MSG_RETURN(MSG_TRANSFER_FAILURE);
  92 }
  93
  94 /** \ingroup msg_gos_functions
  95  * \brief Test whether there is a pending communication on a channel.
  96  *
  97  * It takes one parameter.
  98  * \param channel the channel on which the agent should be
  99    listening. This value has to be >=0 and < than the maximal
 100    number of channels fixed with MSG_set_channel_number().
 101  * \return 1 if there is a pending communication and 0 otherwise
 102  */
 103 int MSG_task_Iprobe(m_channel_t channel)
 104 {
 105   m_host_t h = NULL;
 106   simdata_host_t h_simdata = NULL;
 107
 108   CHECK_HOST();
 109   h = MSG_host_self();
 110   h_simdata = h->simdata;
 111   return(xbt_fifo_getFirstItem(h_simdata->mbox[channel])!=NULL);
 112 }
 113
 114 /** \ingroup msg_gos_functions
 115  * \brief Put a task on a channel of an host and waits for the end of the
 116  * transmission.
 117  *
 118  * This function is used for describing the behavior of an agent. It
 119  * takes three parameter.
 120  * \param task a #m_task_t to send on another location. This task
 121    will not be usable anymore when the function will return. There is
 122    no automatic task duplication and you have to save your parameters
 123    before calling this function. Tasks are unique and once it has been
 124    sent to another location, you should not access it anymore. You do
 125    not need to call MSG_task_destroy() but to avoid using, as an
 126    effect of inattention, this task anymore, you definitely should
 127    renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
 128    can be transfered iff it has been correctly created with
 129    MSG_task_create().
 130  * \param dest the destination of the message
 131  * \param channel the channel on which the agent should put this
 132    task. This value has to be >=0 and < than the maximal number of
 133    channels fixed with MSG_set_channel_number().
 134  * \return #MSG_FATAL if \a task is not properly initialized and
 135  * #MSG_OK otherwise.
 136  */
 137 MSG_error_t MSG_task_put(m_task_t task,
 138                          m_host_t dest, m_channel_t channel)
 139 {
 140   m_process_t process = MSG_process_self();
 141   simdata_task_t task_simdata = NULL;
 142   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
 143   m_host_t local_host = NULL;
 144   m_host_t remote_host = NULL;
 145
 146   CHECK_HOST();
 147
 148   task_simdata = task->simdata;
 149   task_simdata->sender = process;
 150
 151
 152   local_host = ((simdata_process_t) process->simdata)->host;
 153   remote_host = dest;
 154
 155   xbt_fifo_push(((simdata_host_t) remote_host->simdata)->
 156                 mbox[channel], task);
 157
 158   if(remote_host->simdata->sleeping[channel])
 159     MSG_process_resume(remote_host->simdata->sleeping[channel]);
 160   else
 161     MSG_process_suspend(process);
 162
 163   do {
 164     __MSG_task_wait_event(process, task);
 165     state=surf_workstation_resource->common_public->action_get_state(task_simdata->comm);
 166   } while (state==SURF_ACTION_RUNNING);
 167
 168   MSG_task_destroy(task);
 169
 170   if(state == SURF_ACTION_DONE) MSG_RETURN(MSG_OK);
 171   else if(surf_workstation_resource->extension_public->get_state(local_host->simdata->host)
 172           == SURF_CPU_OFF)
 173     MSG_RETURN(MSG_HOST_FAILURE);
 174   else MSG_RETURN(MSG_TRANSFER_FAILURE);
 175 }
 176
 177 /** \ingroup msg_gos_functions
 178  * \brief Executes a task and waits for its termination.
 179  *
 180  * This function is used for describing the behavior of an agent. It
 181  * takes only one parameter.
 182  * \param task a #m_task_t to execute on the location on which the
 183    agent is running.
 184  * \return #MSG_FATAL if \a task is not properly initialized and
 185  * #MSG_OK otherwise.
 186  */
 187 MSG_error_t MSG_task_execute(m_task_t task)
 188 {
 189   m_process_t process = MSG_process_self();
 190
 191   __MSG_task_execute(process, task);
 192   return __MSG_wait_for_computation(process,task);
 193 }
 194
 195 void __MSG_task_execute(m_process_t process, m_task_t task)
 196 {
 197   simdata_task_t simdata = NULL;
 198
 199   CHECK_HOST();
 200
 201   simdata = task->simdata;
 202
 203   simdata->compute = surf_workstation_resource->extension_public->
 204     execute(MSG_process_get_host(process)->simdata->host,
 205             simdata->computation_amount);
 206 }
 207
 208 MSG_error_t __MSG_wait_for_computation(m_process_t process, m_task_t task)
 209 {
 210   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
 211   simdata_task_t simdata = task->simdata;
 212
 213   simdata->using++;
 214   do {
 215     __MSG_task_wait_event(process, task);
 216     state=surf_workstation_resource->common_public->action_get_state(simdata->compute);
 217   } while (state==SURF_ACTION_RUNNING);
 218   simdata->using--;
 219
 220
 221   if(state == SURF_ACTION_DONE) MSG_RETURN(MSG_OK);
 222   else if(surf_workstation_resource->extension_public->
 223           get_state(MSG_process_get_host(process)->simdata->host)
 224           == SURF_CPU_OFF)
 225     MSG_RETURN(MSG_HOST_FAILURE);
 226   else MSG_RETURN(MSG_TRANSFER_FAILURE);
 227 }
 228
 229 /** \ingroup msg_gos_functions
 230  * \brief Sleep for the specified number of seconds
 231  *
 232  * Makes the current process sleep until \a time seconds have elapsed.
 233  *
 234  * \param nb_sec a number of second
 235  */
 236 MSG_error_t MSG_process_sleep(long double nb_sec)
 237 {
 238   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
 239   m_process_t process = MSG_process_self();
 240   m_task_t dummy = NULL;
 241   simdata_task_t simdata = NULL;
 242
 243   CHECK_HOST();
 244   dummy = MSG_task_create("MSG_sleep", nb_sec, 0.0, NULL);
 245   simdata = dummy->simdata;
 246
 247   simdata->compute = surf_workstation_resource->extension_public->
 248     sleep(MSG_process_get_host(process)->simdata->host,
 249             simdata->computation_amount);
 250
 251
 252   simdata->using++;
 253   do {
 254     __MSG_task_wait_event(process, dummy);
 255     state=surf_workstation_resource->common_public->action_get_state(simdata->compute);
 256   } while (state==SURF_ACTION_RUNNING);
 257   simdata->using--;
 258
 259   if(state == SURF_ACTION_DONE) {
 260     if(surf_workstation_resource->extension_public->
 261        get_state(MSG_process_get_host(process)->simdata->host)
 262        == SURF_CPU_OFF)
 263       MSG_RETURN(MSG_HOST_FAILURE);
 264
 265     if(MSG_process_isSuspended(process)) {
 266       MSG_process_suspend(MSG_process_self());
 267     }
 268     if(surf_workstation_resource->extension_public->
 269        get_state(MSG_process_get_host(process)->simdata->host)
 270        == SURF_CPU_OFF)
 271       MSG_RETURN(MSG_HOST_FAILURE);
 272     MSG_task_destroy(dummy);
 273     MSG_RETURN(MSG_OK);
 274   } else MSG_RETURN(MSG_HOST_FAILURE);
 275 }
 276
 277 /** \ingroup msg_gos_functions
 278  * \brief Return the number of MSG tasks currently running on a
 279  * the host of the current running process.
 280  */
 281 int MSG_get_msgload(void)
 282 {
 283   CHECK_HOST();
 284   xbt_assert0(0,"Not implemented yet!");
 285
 286   return 1;
 287 }
 288
 289 /** \ingroup msg_gos_functions
 290  *
 291  * \brief Return the the last value returned by a MSG function (except
 292  * MSG_get_errno...).
 293  */
 294 MSG_error_t MSG_get_errno(void)
 295 {
 296   return PROCESS_GET_ERRNO();
 297 }
 298