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 /** \defgroup msg_gos_functions MSG Operating System Functions
  15  *  \brief This section describes the functions that can be used
  16  *  by an agent for handling some task.
  17  */
  18
  19 /* \ingroup msg_gos_functions
  20  * \brief This function is now deprecated and useless. Please stop using it.
  21  */
  22 MSG_error_t MSG_process_start(m_process_t process)
  23 {
  24   xbt_assert0(0,"This function is now deprecated and useless. Please stop using it.");
  25
  26   return MSG_OK;
  27 }
  28
  29 /** \ingroup msg_gos_functions
  30  * \brief Listen on a channel and wait for receiving a task.
  31  *
  32  * It takes two parameters.
  33  * \param task a memory location for storing a #m_task_t. It will
  34    hold a task when this function will return. Thus \a task should not
  35    be equal to \c NULL and \a *task should be equal to \c NULL. If one of
  36    those two condition does not hold, there will be a warning message.
  37  * \param channel the channel on which the agent should be
  38    listening. This value has to be >=0 and < than the maximal
  39    number of channels fixed with MSG_set_channel_number().
  40  * \return #MSG_FATAL if \a task is equal to \c NULL, #MSG_WARNING
  41  * if \a *task is not equal to \c NULL, and #MSG_OK otherwise.
  42  */
  43 MSG_error_t MSG_task_get(m_task_t * task,
  44                          m_channel_t channel)
  45 {
  46   return MSG_task_get_with_time_out(task, channel, -1);
  47 }
  48
  49 /** \ingroup msg_gos_functions
  50  * \brief Listen on a channel and wait for receiving a task with a timeout.
  51  *
  52  * It takes three parameters.
  53  * \param task a memory location for storing a #m_task_t. It will
  54    hold a task when this function will return. Thus \a task should not
  55    be equal to \c NULL and \a *task should be equal to \c NULL. If one of
  56    those two condition does not hold, there will be a warning message.
  57  * \param channel the channel on which the agent should be
  58    listening. This value has to be >=0 and < than the maximal
  59    number of channels fixed with MSG_set_channel_number().
  60  * \param max_duration the maximum time to wait for a task before giving
  61     up. In such a case, \a task will not be modified and will still be
  62     equal to \c NULL when returning.
  63  * \return #MSG_FATAL if \a task is equal to \c NULL, #MSG_WARNING
  64    if \a *task is not equal to \c NULL, and #MSG_OK otherwise.
  65  */
  66
  67 MSG_error_t MSG_task_get_with_time_out(m_task_t * task,
  68                                        m_channel_t channel,
  69                                        double max_duration)
  70 {
  71   m_process_t process = MSG_process_self();
  72   m_task_t t = NULL;
  73   m_host_t h = NULL;
  74   simdata_task_t t_simdata = NULL;
  75   simdata_host_t h_simdata = NULL;
  76   int warning = 0;
  77   int first_time = 1;
  78   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
  79
  80   CHECK_HOST();
  81   /* Sanity check */
  82   xbt_assert0(task,"Null pointer for the task\n");
  83
  84   if (*task)
  85     CRITICAL0("MSG_task_get() was asked to write in a non empty task struct.");
  86
  87   /* Get the task */
  88   h = MSG_host_self();
  89   h_simdata = h->simdata;
  90
  91   DEBUG2("Waiting for a task on channel %d (%s)", channel,h->name);
  92
  93   while ((t = xbt_fifo_shift(h_simdata->mbox[channel])) == NULL) {
  94     if(max_duration>0) {
  95       if(!first_time) {
  96         MSG_RETURN(MSG_OK);
  97       }
  98     }
  99     xbt_assert2(!(h_simdata->sleeping[channel]),
 100                 "A process (%s(%d)) is already blocked on this channel",
 101                 h_simdata->sleeping[channel]->name,
 102                 h_simdata->sleeping[channel]->simdata->PID);
 103     h_simdata->sleeping[channel] = process; /* I'm waiting. Wake me up when you're ready */
 104     if(max_duration>0) {
 105       __MSG_process_block(max_duration);
 106     } else {
 107       __MSG_process_block(-1);
 108     }
 109     if(surf_workstation_resource->extension_public->get_state(h_simdata->host)
 110        == SURF_CPU_OFF)
 111       MSG_RETURN(MSG_HOST_FAILURE);
 112     h_simdata->sleeping[channel] = NULL;
 113     first_time = 0;
 114     /* OK, we should both be ready now. Are you there ? */
 115   }
 116
 117   t_simdata = t->simdata;
 118   /*   *task = __MSG_task_copy(t); */
 119   *task=t;
 120
 121   /* Transfer */
 122   t_simdata->using++;
 123
 124   t_simdata->comm = surf_workstation_resource->extension_public->
 125     communicate(MSG_process_get_host(t_simdata->sender)->simdata->host,
 126                 h->simdata->host, t_simdata->message_size,t_simdata->rate);
 127
 128   surf_workstation_resource->common_public->action_set_data(t_simdata->comm,t);
 129
 130   if(__MSG_process_isBlocked(t_simdata->sender))
 131     __MSG_process_unblock(t_simdata->sender);
 132
 133   PAJE_PROCESS_PUSH_STATE(process,"C");
 134
 135   do {
 136     __MSG_task_wait_event(process, t);
 137     state=surf_workstation_resource->common_public->action_get_state(t_simdata->comm);
 138   } while (state==SURF_ACTION_RUNNING);
 139
 140   if(t->simdata->using>1) {
 141     xbt_fifo_unshift(msg_global->process_to_run,process);
 142     xbt_context_yield();
 143   }
 144
 145   PAJE_PROCESS_POP_STATE(process);
 146   PAJE_COMM_STOP(process,t,channel);
 147
 148   if(state == SURF_ACTION_DONE) {
 149     if(surf_workstation_resource->common_public->action_free(t_simdata->comm))
 150       t_simdata->comm = NULL;
 151     MSG_RETURN(MSG_OK);
 152   } else if(surf_workstation_resource->extension_public->get_state(h_simdata->host)
 153           == SURF_CPU_OFF) {
 154     if(surf_workstation_resource->common_public->action_free(t_simdata->comm))
 155       t_simdata->comm = NULL;
 156     MSG_RETURN(MSG_HOST_FAILURE);
 157   } else {
 158     if(surf_workstation_resource->common_public->action_free(t_simdata->comm))
 159       t_simdata->comm = NULL;
 160     MSG_RETURN(MSG_TRANSFER_FAILURE);
 161   }
 162 }
 163
 164 /** \ingroup msg_gos_functions
 165  * \brief Test whether there is a pending communication on a channel.
 166  *
 167  * It takes one parameter.
 168  * \param channel the channel on which the agent should be
 169    listening. This value has to be >=0 and < than the maximal
 170    number of channels fixed with MSG_set_channel_number().
 171  * \return 1 if there is a pending communication and 0 otherwise
 172  */
 173 int MSG_task_Iprobe(m_channel_t channel)
 174 {
 175   m_host_t h = NULL;
 176   simdata_host_t h_simdata = NULL;
 177
 178   DEBUG2("Probing on channel %d (%s)", channel,h->name);
 179   CHECK_HOST();
 180   h = MSG_host_self();
 181   h_simdata = h->simdata;
 182   return(xbt_fifo_getFirstItem(h_simdata->mbox[channel])!=NULL);
 183 }
 184
 185 /** \ingroup msg_gos_functions
 186  * \brief Test whether there is a pending communication on a channel, and who sent it.
 187  *
 188  * It takes one parameter.
 189  * \param channel the channel on which the agent should be
 190    listening. This value has to be >=0 and < than the maximal
 191    number of channels fixed with MSG_set_channel_number().
 192  * \return -1 if there is no pending communication and the PID of the process who sent it otherwise
 193  */
 194 int MSG_task_probe_from(m_channel_t channel)
 195 {
 196   m_host_t h = NULL;
 197   simdata_host_t h_simdata = NULL;
 198   xbt_fifo_item_t item;
 199   m_task_t t;
 200
 201   CHECK_HOST();
 202   h = MSG_host_self();
 203   h_simdata = h->simdata;
 204
 205   DEBUG2("Probing on channel %d (%s)", channel,h->name);
 206
 207   item = xbt_fifo_getFirstItem(h->simdata->mbox[channel]);
 208   if (!item || !(t = xbt_fifo_get_item_content(item)))
 209     return -1;
 210
 211   return MSG_process_get_PID(t->simdata->sender);
 212 }
 213
 214 MSG_error_t MSG_channel_select_from(m_channel_t channel, double max_duration,
 215                                     int *PID)
 216 {
 217   m_host_t h = NULL;
 218   simdata_host_t h_simdata = NULL;
 219   xbt_fifo_item_t item;
 220   m_task_t t;
 221   int first_time = 1;
 222   m_process_t process = MSG_process_self();
 223
 224   if(PID) {
 225     *PID = -1;
 226   }
 227
 228   if(max_duration==0.0) {
 229     return MSG_task_probe_from(channel);
 230   } else {
 231     CHECK_HOST();
 232     h = MSG_host_self();
 233     h_simdata = h->simdata;
 234
 235     DEBUG2("Probing on channel %d (%s)", channel,h->name);
 236     while(!(item = xbt_fifo_getFirstItem(h->simdata->mbox[channel]))) {
 237       if(max_duration>0) {
 238         if(!first_time) {
 239           MSG_RETURN(MSG_OK);
 240         }
 241       }
 242       xbt_assert2(!(h_simdata->sleeping[channel]),
 243                   "A process (%s(%d)) is already blocked on this channel",
 244                   h_simdata->sleeping[channel]->name,
 245                   h_simdata->sleeping[channel]->simdata->PID);
 246       h_simdata->sleeping[channel] = process; /* I'm waiting. Wake me up when you're ready */
 247       if(max_duration>0) {
 248         __MSG_process_block(max_duration);
 249       } else {
 250         __MSG_process_block(-1);
 251       }
 252       if(surf_workstation_resource->extension_public->get_state(h_simdata->host)
 253          == SURF_CPU_OFF) {
 254         MSG_RETURN(MSG_HOST_FAILURE);
 255       }
 256       h_simdata->sleeping[channel] = NULL;
 257       first_time = 0;
 258     }
 259     if (!item || !(t = xbt_fifo_get_item_content(item))) {
 260       MSG_RETURN(MSG_OK);
 261     }
 262     if(PID) {
 263       *PID = MSG_process_get_PID(t->simdata->sender);
 264     }
 265     MSG_RETURN(MSG_OK);
 266   }
 267 }
 268 /** \ingroup msg_gos_functions
 269  * \brief Put a task on a channel of an host and waits for the end of the
 270  * transmission.
 271  *
 272  * This function is used for describing the behavior of an agent. It
 273  * takes three parameter.
 274  * \param task a #m_task_t to send on another location. This task
 275    will not be usable anymore when the function will return. There is
 276    no automatic task duplication and you have to save your parameters
 277    before calling this function. Tasks are unique and once it has been
 278    sent to another location, you should not access it anymore. You do
 279    not need to call MSG_task_destroy() but to avoid using, as an
 280    effect of inattention, this task anymore, you definitely should
 281    renitialize it with #MSG_TASK_UNINITIALIZED. Note that this task
 282    can be transfered iff it has been correctly created with
 283    MSG_task_create().
 284  * \param dest the destination of the message
 285  * \param channel the channel on which the agent should put this
 286    task. This value has to be >=0 and < than the maximal number of
 287    channels fixed with MSG_set_channel_number().
 288  * \return #MSG_FATAL if \a task is not properly initialized and
 289  * #MSG_OK otherwise.
 290  */
 291 MSG_error_t MSG_task_put(m_task_t task,
 292                          m_host_t dest, m_channel_t channel)
 293 {
 294   m_process_t process = MSG_process_self();
 295   simdata_task_t task_simdata = NULL;
 296   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
 297   m_host_t local_host = NULL;
 298   m_host_t remote_host = NULL;
 299
 300   CHECK_HOST();
 301
 302   task_simdata = task->simdata;
 303   task_simdata->sender = process;
 304   xbt_assert0(task_simdata->using==1,"Gargl!");
 305   task_simdata->comm = NULL;
 306
 307   local_host = ((simdata_process_t) process->simdata)->host;
 308   remote_host = dest;
 309
 310   DEBUG4("Trying to send a task (%g Mb) from %s to %s on channel %d",
 311          task->simdata->message_size,local_host->name, remote_host->name, channel);
 312
 313   xbt_fifo_push(((simdata_host_t) remote_host->simdata)->
 314                 mbox[channel], task);
 315
 316   PAJE_COMM_START(process,task,channel);
 317
 318   if(remote_host->simdata->sleeping[channel])
 319     __MSG_process_unblock(remote_host->simdata->sleeping[channel]);
 320
 321   process->simdata->put_host = dest;
 322   process->simdata->put_channel = channel;
 323   while(!(task_simdata->comm))
 324     __MSG_process_block(-1);
 325   surf_workstation_resource->common_public->action_use(task_simdata->comm);
 326   process->simdata->put_host = NULL;
 327   process->simdata->put_channel = -1;
 328
 329
 330   PAJE_PROCESS_PUSH_STATE(process,"C");
 331
 332   state=surf_workstation_resource->common_public->action_get_state(task_simdata->comm);
 333   while (state==SURF_ACTION_RUNNING) {
 334     __MSG_task_wait_event(process, task);
 335     state=surf_workstation_resource->common_public->action_get_state(task_simdata->comm);
 336   }
 337
 338
 339   PAJE_PROCESS_POP_STATE(process);
 340
 341   if(state == SURF_ACTION_DONE) {
 342     if(surf_workstation_resource->common_public->action_free(task_simdata->comm))
 343       task_simdata->comm = NULL;
 344     MSG_task_destroy(task);
 345     MSG_RETURN(MSG_OK);
 346   } else if(surf_workstation_resource->extension_public->get_state(local_host->simdata->host)
 347             == SURF_CPU_OFF) {
 348     if(surf_workstation_resource->common_public->action_free(task_simdata->comm))
 349       task_simdata->comm = NULL;
 350     MSG_task_destroy(task);
 351     MSG_RETURN(MSG_HOST_FAILURE);
 352   } else {
 353     if(surf_workstation_resource->common_public->action_free(task_simdata->comm))
 354       task_simdata->comm = NULL;
 355     MSG_task_destroy(task);
 356     MSG_RETURN(MSG_TRANSFER_FAILURE);
 357   }
 358 }
 359
 360 /** \ingroup msg_gos_functions
 361  * \brief Does exactly the same as MSG_task_put but with a bounded transmition
 362  * rate.
 363  *
 364  * \sa MSG_task_put
 365  */
 366 MSG_error_t MSG_task_put_bounded(m_task_t task,
 367                                  m_host_t dest, m_channel_t channel,
 368                                  double max_rate)
 369 {
 370   task->simdata->rate=max_rate;
 371   return(MSG_task_put(task, dest, channel));
 372   task->simdata->rate=-1.0;
 373 }
 374
 375 /** \ingroup msg_gos_functions
 376  * \brief Executes a task and waits for its termination.
 377  *
 378  * This function is used for describing the behavior of an agent. It
 379  * takes only one parameter.
 380  * \param task a #m_task_t to execute on the location on which the
 381    agent is running.
 382  * \return #MSG_FATAL if \a task is not properly initialized and
 383  * #MSG_OK otherwise.
 384  */
 385 MSG_error_t MSG_task_execute(m_task_t task)
 386 {
 387   m_process_t process = MSG_process_self();
 388   MSG_error_t res;
 389
 390   DEBUG1("Computing on %s", process->simdata->host->name);
 391
 392   __MSG_task_execute(process, task);
 393
 394   PAJE_PROCESS_PUSH_STATE(process,"E");
 395   res = __MSG_wait_for_computation(process,task);
 396   PAJE_PROCESS_POP_STATE(process);
 397   return res;
 398 }
 399
 400 void __MSG_task_execute(m_process_t process, m_task_t task)
 401 {
 402   simdata_task_t simdata = NULL;
 403
 404   CHECK_HOST();
 405
 406   simdata = task->simdata;
 407
 408   simdata->compute = surf_workstation_resource->extension_public->
 409     execute(MSG_process_get_host(process)->simdata->host,
 410             simdata->computation_amount);
 411   surf_workstation_resource->common_public->action_set_data(simdata->compute,task);
 412 }
 413
 414 MSG_error_t __MSG_wait_for_computation(m_process_t process, m_task_t task)
 415 {
 416   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
 417   simdata_task_t simdata = task->simdata;
 418
 419   simdata->using++;
 420   do {
 421     __MSG_task_wait_event(process, task);
 422     state=surf_workstation_resource->common_public->action_get_state(simdata->compute);
 423   } while (state==SURF_ACTION_RUNNING);
 424   simdata->using--;
 425
 426
 427   if(state == SURF_ACTION_DONE) {
 428     if(surf_workstation_resource->common_public->action_free(simdata->compute))
 429       simdata->compute = NULL;
 430     MSG_RETURN(MSG_OK);
 431   } else if(surf_workstation_resource->extension_public->
 432             get_state(MSG_process_get_host(process)->simdata->host)
 433             == SURF_CPU_OFF) {
 434     if(surf_workstation_resource->common_public->action_free(simdata->compute))
 435       simdata->compute = NULL;
 436     MSG_RETURN(MSG_HOST_FAILURE);
 437   } else {
 438     if(surf_workstation_resource->common_public->action_free(simdata->compute))
 439       simdata->compute = NULL;
 440     MSG_RETURN(MSG_TRANSFER_FAILURE);
 441   }
 442 }
 443
 444 m_task_t MSG_parallel_task_create(const char *name,
 445                                   int host_nb,
 446                                   const m_host_t *host_list,
 447                                   double *computation_amount,
 448                                   double *communication_amount,
 449                                   void *data)
 450 {
 451   simdata_task_t simdata = xbt_new0(s_simdata_task_t,1);
 452   m_task_t task = xbt_new0(s_m_task_t,1);
 453   int i;
 454
 455   /* Task structure */
 456   task->name = xbt_strdup(name);
 457   task->simdata = simdata;
 458   task->data = data;
 459
 460   /* Simulator Data */
 461   simdata->sleeping = xbt_dynar_new(sizeof(m_process_t),NULL);
 462   simdata->rate = -1.0;
 463   simdata->using = 1;
 464   simdata->sender = NULL;
 465   simdata->host_nb = host_nb;
 466
 467   simdata->host_list = xbt_new0(void *, host_nb);
 468   simdata->comp_amount = computation_amount;
 469   simdata->comm_amount = communication_amount;
 470
 471   for(i=0;i<host_nb;i++)
 472     simdata->host_list[i] = host_list[i]->simdata->host;
 473
 474   return task;
 475 }
 476
 477
 478 static void __MSG_parallel_task_execute(m_process_t process, m_task_t task)
 479 {
 480   simdata_task_t simdata = NULL;
 481
 482   CHECK_HOST();
 483
 484   simdata = task->simdata;
 485
 486   xbt_assert0(simdata->host_nb,"This is not a parallel task. Go to hell.");
 487
 488   simdata->compute = surf_workstation_resource->extension_public->
 489   execute_parallel_task(task->simdata->host_nb,
 490                         task->simdata->host_list,
 491                         task->simdata->comp_amount,
 492                         task->simdata->comm_amount,
 493                         1.0,
 494                         -1.0);
 495   if(simdata->compute)
 496     surf_workstation_resource->common_public->action_set_data(simdata->compute,task);
 497 }
 498
 499 MSG_error_t MSG_parallel_task_execute(m_task_t task)
 500 {
 501   m_process_t process = MSG_process_self();
 502   MSG_error_t res;
 503
 504   DEBUG0("Computing on a tons of guys");
 505
 506   __MSG_parallel_task_execute(process, task);
 507
 508   if(task->simdata->compute)
 509     res = __MSG_wait_for_computation(process,task);
 510   else
 511     res = MSG_OK;
 512
 513   return res;
 514 }
 515
 516
 517 /** \ingroup msg_gos_functions
 518  * \brief Sleep for the specified number of seconds
 519  *
 520  * Makes the current process sleep until \a time seconds have elapsed.
 521  *
 522  * \param nb_sec a number of second
 523  */
 524 MSG_error_t MSG_process_sleep(double nb_sec)
 525 {
 526   e_surf_action_state_t state = SURF_ACTION_NOT_IN_THE_SYSTEM;
 527   m_process_t process = MSG_process_self();
 528   m_task_t dummy = NULL;
 529   simdata_task_t simdata = NULL;
 530
 531   CHECK_HOST();
 532   dummy = MSG_task_create("MSG_sleep", nb_sec, 0.0, NULL);
 533   simdata = dummy->simdata;
 534
 535   simdata->compute = surf_workstation_resource->extension_public->
 536     sleep(MSG_process_get_host(process)->simdata->host,
 537             simdata->computation_amount);
 538   surf_workstation_resource->common_public->action_set_data(simdata->compute,dummy);
 539
 540
 541   simdata->using++;
 542   do {
 543     __MSG_task_wait_event(process, dummy);
 544     state=surf_workstation_resource->common_public->action_get_state(simdata->compute);
 545   } while (state==SURF_ACTION_RUNNING);
 546   simdata->using--;
 547
 548   if(state == SURF_ACTION_DONE) {
 549     if(surf_workstation_resource->extension_public->
 550        get_state(MSG_process_get_host(process)->simdata->host)
 551        == SURF_CPU_OFF) {
 552       if(surf_workstation_resource->common_public->action_free(simdata->compute))
 553         simdata->compute = NULL;
 554       MSG_RETURN(MSG_HOST_FAILURE);
 555     }
 556     if(__MSG_process_isBlocked(process)) {
 557       __MSG_process_unblock(MSG_process_self());
 558     }
 559     if(surf_workstation_resource->extension_public->
 560        get_state(MSG_process_get_host(process)->simdata->host)
 561        == SURF_CPU_OFF) {
 562       if(surf_workstation_resource->common_public->action_free(simdata->compute))
 563         simdata->compute = NULL;
 564       MSG_RETURN(MSG_HOST_FAILURE);
 565     }
 566     if(surf_workstation_resource->common_public->action_free(simdata->compute))
 567       simdata->compute = NULL;
 568     MSG_task_destroy(dummy);
 569     MSG_RETURN(MSG_OK);
 570   } else MSG_RETURN(MSG_HOST_FAILURE);
 571 }
 572
 573 /** \ingroup msg_gos_functions
 574  * \brief Return the number of MSG tasks currently running on a
 575  * the host of the current running process.
 576  */
 577 static int MSG_get_msgload(void)
 578 {
 579   m_process_t process;
 580
 581   CHECK_HOST();
 582
 583   xbt_assert0(0, "This function is still to be specified correctly (what do you mean by 'load', exactly?). In the meantime, please don't use it");
 584   process = MSG_process_self();
 585   return xbt_fifo_size(process->simdata->host->simdata->process_list);
 586 }
 587
 588 /** \ingroup msg_gos_functions
 589  *
 590  * \brief Return the the last value returned by a MSG function (except
 591  * MSG_get_errno...).
 592  */
 593 MSG_error_t MSG_get_errno(void)
 594 {
 595   return PROCESS_GET_ERRNO();
 596 }