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_block();
  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
  79   t_simdata->comm = surf_workstation_resource->extension_public->
  80     communicate(MSG_process_get_host(t_simdata->sender)->simdata->host,
  81                 h->simdata->host, t_simdata->message_size,t_simdata->rate);
  82
  83   surf_workstation_resource->common_public->action_set_data(t_simdata->comm,t);
  84
  85   if(__MSG_process_isBlocked(t_simdata->sender))
  86     __MSG_process_unblock(t_simdata->sender);
  87
  88   do {
  89     __MSG_task_wait_event(process, t);
  90     state=surf_workstation_resource->common_public->action_get_state(t_simdata->comm);
  91   } while (state==SURF_ACTION_RUNNING);
  92
  93   if(state == SURF_ACTION_DONE) MSG_RETURN(MSG_OK);
  94   else if(surf_workstation_resource->extension_public->get_state(h_simdata->host)
  95           == SURF_CPU_OFF)
  96     MSG_RETURN(MSG_HOST_FAILURE);
  97   else MSG_RETURN(MSG_TRANSFER_FAILURE);
  98 }
  99
 100 /** \ingroup msg_gos_functions
 101  * \brief Test whether there is a pending communication on a channel.
 102  *
 103  * It takes one parameter.
 104  * \param channel the channel on which the agent should be
 105    listening. This value has to be >=0 and < than the maximal
 106    number of channels fixed with MSG_set_channel_number().
 107  * \return 1 if there is a pending communication and 0 otherwise
 108  */
 109 int MSG_task_Iprobe(m_channel_t channel)
 110 {
 111   m_host_t h = NULL;
 112   simdata_host_t h_simdata = NULL;
 113
 114   CHECK_HOST();
 115   h = MSG_host_self();
 116   h_simdata = h->simdata;
 117   return(xbt_fifo_getFirstItem(h_simdata->mbox[channel])!=NULL);
 118 }
 119
 120 /** \ingroup msg_gos_functions
 121  * \brief Test whether there is a pending communication on a channel, and who sent it.
 122  *
 123  * It takes one parameter.
 124  * \param channel the channel on which the agent should be
 125    listening. This value has to be >=0 and < than the maximal
 126    number of channels fixed with MSG_set_channel_number().
 127  * \return -1 if there is no pending communication and the PID of the process who sent it otherwise
 128  */
 129 int MSG_task_probe_from(m_channel_t channel)
 130 {
 131   m_host_t h = NULL;
 132   simdata_host_t h_simdata = NULL;
 133   xbt_fifo_item_t item;
 134   m_task_t t;
 135
 136   CHECK_HOST();
 137   h = MSG_host_self();
 138   h_simdata = h->simdata;
 139
 140   item = xbt_fifo_getFirstItem(((simdata_host_t)h->simdata)->mbox[channel]);
 141   if (!item || !(t = xbt_fifo_get_item_content(item)) || (simdata_task_t)t->simdata)
 142     return -1;
 143
 144   return MSG_process_get_PID(((simdata_task_t)t->simdata)->sender);
 145 }
 146
 147 /** \ingroup msg_gos_functions
 148  * \brief Put a task on a channel of an host and waits for the end of the
 149  * transmission.
 150  *
 151  * This function is used for describing the behavior of an agent. It
 152  * takes three parameter.
 153  * \param task a #m_task_t to send on another location. This task
 154    will not be usable anymore when the function will return. There is
 155    no automatic task duplication and you have to save your parameters
 156    before calling this function. Tasks are unique and once it has been
 157    sent to another location, you should not access it anymore. You do
 158    not need to call MSG_task_destroy() but to avoid using, as an
 159    effect of inattention, this task anymore, you definitely should
 160    renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
 161    can be transfered iff it has been correctly created with
 162    MSG_task_create().
 163  * \param dest the destination of the message
 164  * \param channel the channel on which the agent should put this
 165    task. This value has to be >=0 and < than the maximal number of
 166    channels fixed with MSG_set_channel_number().
 167  * \return #MSG_FATAL if \a task is not properly initialized and
 168  * #MSG_OK otherwise.
 169  */
 170 MSG_error_t MSG_task_put(m_task_t task,
 171                          m_host_t dest, m_channel_t channel)
 172 {
 173   m_process_t process = MSG_process_self();
 174   simdata_task_t task_simdata = NULL;
 175   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
 176   m_host_t local_host = NULL;
 177   m_host_t remote_host = NULL;
 178
 179   CHECK_HOST();
 180
 181   task_simdata = task->simdata;
 182   task_simdata->sender = process;
 183   xbt_assert0(task_simdata->using==1,"Gargl!");
 184   task_simdata->comm = NULL;
 185
 186   local_host = ((simdata_process_t) process->simdata)->host;
 187   remote_host = dest;
 188
 189   xbt_fifo_push(((simdata_host_t) remote_host->simdata)->
 190                 mbox[channel], task);
 191
 192   if(remote_host->simdata->sleeping[channel])
 193     __MSG_process_unblock(remote_host->simdata->sleeping[channel]);
 194 /*   else { */
 195     process->simdata->put_host = dest;
 196     process->simdata->put_channel = channel;
 197     while(!(task_simdata->comm))
 198       __MSG_process_block();
 199     process->simdata->put_host = NULL;
 200     process->simdata->put_channel = -1;
 201 /*   } */
 202
 203   do {
 204     __MSG_task_wait_event(process, task);
 205     state=surf_workstation_resource->common_public->action_get_state(task_simdata->comm);
 206   } while (state==SURF_ACTION_RUNNING);
 207
 208   MSG_task_destroy(task);
 209
 210   if(state == SURF_ACTION_DONE) MSG_RETURN(MSG_OK);
 211   else if(surf_workstation_resource->extension_public->get_state(local_host->simdata->host)
 212           == SURF_CPU_OFF)
 213     MSG_RETURN(MSG_HOST_FAILURE);
 214   else MSG_RETURN(MSG_TRANSFER_FAILURE);
 215 }
 216
 217 MSG_error_t MSG_task_put_bounded(m_task_t task,
 218                                  m_host_t dest, m_channel_t channel,
 219                                  long double max_rate)
 220 {
 221   task->simdata->rate=max_rate;
 222   return(MSG_task_put(task, dest, channel));
 223 }
 224
 225 /** \ingroup msg_gos_functions
 226  * \brief Executes a task and waits for its termination.
 227  *
 228  * This function is used for describing the behavior of an agent. It
 229  * takes only one parameter.
 230  * \param task a #m_task_t to execute on the location on which the
 231    agent is running.
 232  * \return #MSG_FATAL if \a task is not properly initialized and
 233  * #MSG_OK otherwise.
 234  */
 235 MSG_error_t MSG_task_execute(m_task_t task)
 236 {
 237   m_process_t process = MSG_process_self();
 238
 239   __MSG_task_execute(process, task);
 240   return __MSG_wait_for_computation(process,task);
 241 }
 242
 243 void __MSG_task_execute(m_process_t process, m_task_t task)
 244 {
 245   simdata_task_t simdata = NULL;
 246
 247   CHECK_HOST();
 248
 249   simdata = task->simdata;
 250
 251   simdata->compute = surf_workstation_resource->extension_public->
 252     execute(MSG_process_get_host(process)->simdata->host,
 253             simdata->computation_amount);
 254   surf_workstation_resource->common_public->action_set_data(simdata->compute,task);
 255 }
 256
 257 MSG_error_t __MSG_wait_for_computation(m_process_t process, m_task_t task)
 258 {
 259   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
 260   simdata_task_t simdata = task->simdata;
 261
 262   simdata->using++;
 263   do {
 264     __MSG_task_wait_event(process, task);
 265     state=surf_workstation_resource->common_public->action_get_state(simdata->compute);
 266   } while (state==SURF_ACTION_RUNNING);
 267   simdata->using--;
 268
 269
 270   if(state == SURF_ACTION_DONE) MSG_RETURN(MSG_OK);
 271   else if(surf_workstation_resource->extension_public->
 272           get_state(MSG_process_get_host(process)->simdata->host)
 273           == SURF_CPU_OFF)
 274     MSG_RETURN(MSG_HOST_FAILURE);
 275   else MSG_RETURN(MSG_TRANSFER_FAILURE);
 276 }
 277
 278 /** \ingroup msg_gos_functions
 279  * \brief Sleep for the specified number of seconds
 280  *
 281  * Makes the current process sleep until \a time seconds have elapsed.
 282  *
 283  * \param nb_sec a number of second
 284  */
 285 MSG_error_t MSG_process_sleep(long double nb_sec)
 286 {
 287   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
 288   m_process_t process = MSG_process_self();
 289   m_task_t dummy = NULL;
 290   simdata_task_t simdata = NULL;
 291
 292   CHECK_HOST();
 293   dummy = MSG_task_create("MSG_sleep", nb_sec, 0.0, NULL);
 294   simdata = dummy->simdata;
 295
 296   simdata->compute = surf_workstation_resource->extension_public->
 297     sleep(MSG_process_get_host(process)->simdata->host,
 298             simdata->computation_amount);
 299   surf_workstation_resource->common_public->action_set_data(simdata->compute,dummy);
 300
 301
 302   simdata->using++;
 303   do {
 304     __MSG_task_wait_event(process, dummy);
 305     state=surf_workstation_resource->common_public->action_get_state(simdata->compute);
 306   } while (state==SURF_ACTION_RUNNING);
 307   simdata->using--;
 308
 309   if(state == SURF_ACTION_DONE) {
 310     if(surf_workstation_resource->extension_public->
 311        get_state(MSG_process_get_host(process)->simdata->host)
 312        == SURF_CPU_OFF)
 313       MSG_RETURN(MSG_HOST_FAILURE);
 314
 315     if(__MSG_process_isBlocked(process)) {
 316       __MSG_process_unblock(MSG_process_self());
 317     }
 318     if(surf_workstation_resource->extension_public->
 319        get_state(MSG_process_get_host(process)->simdata->host)
 320        == SURF_CPU_OFF)
 321       MSG_RETURN(MSG_HOST_FAILURE);
 322     MSG_task_destroy(dummy);
 323     MSG_RETURN(MSG_OK);
 324   } else MSG_RETURN(MSG_HOST_FAILURE);
 325 }
 326
 327 /** \ingroup msg_gos_functions
 328  * \brief Return the number of MSG tasks currently running on a
 329  * the host of the current running process.
 330  */
 331 int MSG_get_msgload(void)
 332 {
 333   CHECK_HOST();
 334   xbt_assert0(0,"Not implemented yet!");
 335
 336   return 1;
 337 }
 338
 339 /** \ingroup msg_gos_functions
 340  *
 341  * \brief Return the the last value returned by a MSG function (except
 342  * MSG_get_errno...).
 343  */
 344 MSG_error_t MSG_get_errno(void)
 345 {
 346   return PROCESS_GET_ERRNO();
 347 }
 348