src/msg/global.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(global, msg,
  12                                 "Logging specific to MSG (global)");
  13
  14 int __stop_at_time = -1.0 ;
  15
  16 MSG_Global_t msg_global = NULL;
  17
  18 /* static void MarkAsFailed(m_task_t t, TBX_HashTable_t failedProcessList); */
  19 /* static xbt_fifo_t MSG_buildFailedHostList(double a, double b); */
  20
  21 /** \defgroup msg_simulation   MSG simulation Functions
  22  *  \brief This section describes the functions you need to know to
  23  *  set up a simulation. You should have a look at \ref MSG_examples
  24  *  to have an overview of their usage.
  25  */
  26
  27 /********************************* MSG **************************************/
  28
  29 /** \ingroup msg_simulation
  30  * \brief Initialize some MSG internal data.
  31  */
  32 void MSG_global_init(void)
  33 {
  34   int argc=0;
  35   char **argv=NULL;
  36
  37   CRITICAL0("Function MSG_global_init() is deprecated by MSG_global_init_args().");
  38   MSG_global_init_args(&argc,argv);
  39 }
  40
  41 /** \ingroup msg_simulation
  42  * \brief Initialize some MSG internal data.
  43  */
  44 void MSG_global_init_args(int *argc, char **argv)
  45 {
  46   if (!msg_global) {
  47     surf_init(argc, argv);      /* Initialize some common structures. Warning, it sets msg_global=NULL */
  48
  49     msg_global = xbt_new0(s_MSG_Global_t,1);
  50
  51     xbt_context_init();
  52     msg_global->host = xbt_fifo_new();
  53     msg_global->process_to_run = xbt_fifo_new();
  54     msg_global->process_list = xbt_fifo_new();
  55     msg_global->max_channel = 0;
  56     msg_global->current_process = NULL;
  57     msg_global->registered_functions = xbt_dict_new();
  58     msg_global->PID = 1;
  59   }
  60 }
  61
  62 /** \ingroup msg_easier_life
  63  * \brief Traces MSG events in the Paje format.
  64  */
  65 void MSG_paje_output(const char *filename)
  66 {
  67   int i;
  68   const char *paje_preembule="%EventDef SetLimits       0\n"
  69     "%  StartTime       date\n"
  70     "%  EndTime date\n"
  71     "%EndEventDef\n"
  72     "%EventDef  PajeDefineContainerType 1\n"
  73     "%  NewType string\n"
  74     "%  ContainerType   string\n"
  75     "%  NewName string\n"
  76     "%EndEventDef\n"
  77     "%EventDef  PajeDefineEventType     2\n"
  78     "%  NewType string\n"
  79     "%  ContainerType   string\n"
  80     "%  NewName string\n"
  81     "%EndEventDef\n"
  82     "%EventDef  PajeDefineStateType     3\n"
  83     "%  NewType string\n"
  84     "%  ContainerType   string\n"
  85     "%  NewName string\n"
  86     "%EndEventDef\n"
  87     "%EventDef  PajeDefineVariableType  4\n"
  88     "%  NewType string\n"
  89     "%  ContainerType   string\n"
  90     "%  NewName string\n"
  91     "%EndEventDef\n"
  92     "%EventDef  PajeDefineLinkType      5\n"
  93     "%  NewType string\n"
  94     "%  ContainerType   string\n"
  95     "%  SourceContainerType     string\n"
  96     "%  DestContainerType       string\n"
  97     "%  NewName string\n"
  98     "%EndEventDef\n"
  99     "%EventDef  PajeDefineEntityValue   6\n"
 100     "%  NewValue        string\n"
 101     "%  EntityType      string\n"
 102     "%  NewName string\n"
 103     "%EndEventDef\n"
 104     "%EventDef  PajeCreateContainer     7\n"
 105     "%  Time    date\n"
 106     "%  NewContainer    string\n"
 107     "%  NewContainerType        string\n"
 108     "%  Container       string\n"
 109     "%  NewName string\n"
 110     "%EndEventDef\n"
 111     "%EventDef  PajeDestroyContainer    8\n"
 112     "%  Time    date\n"
 113     "%  Name    string\n"
 114     "%  Type    string\n"
 115     "%EndEventDef\n"
 116     "%EventDef  PajeNewEvent    9\n"
 117     "%  Time    date\n"
 118     "%  EntityType      string\n"
 119     "%  Container       string\n"
 120     "%  Value   string\n"
 121     "%EndEventDef\n"
 122     "%EventDef  PajeSetState    10\n"
 123     "%  Time    date\n"
 124     "%  EntityType      string\n"
 125     "%  Container       string\n"
 126     "%  Value   string\n"
 127     "%EndEventDef\n"
 128     "%EventDef  PajeSetState    101\n"
 129     "%  Time    date\n"
 130     "%  EntityType      string\n"
 131     "%  Container       string\n"
 132     "%  Value   string\n"
 133     "%  FileName        string\n"
 134     "%  LineNumber      int\n"
 135     "%EndEventDef\n"
 136     "%EventDef  PajePushState   111\n"
 137     "%  Time    date\n"
 138     "%  EntityType      string\n"
 139     "%  Container       string\n"
 140     "%  Value   string\n"
 141     "%  FileName        string\n"
 142     "%  LineNumber      int\n"
 143     "%EndEventDef\n"
 144     "%EventDef  PajePushState   11\n"
 145     "%  Time    date\n"
 146     "%  EntityType      string\n"
 147     "%  Container       string\n"
 148     "%  Value   string\n"
 149     "%EndEventDef\n"
 150     "%EventDef  PajePopState    12\n"
 151     "%  Time    date\n"
 152     "%  EntityType      string\n"
 153     "%  Container       string\n"
 154     "%EndEventDef\n"
 155     "%EventDef  PajeSetVariable 13\n"
 156     "%  Time    date\n"
 157     "%  EntityType      string\n"
 158     "%  Container       string\n"
 159     "%  Value   double\n"
 160     "%EndEventDef\n"
 161     "%EventDef  PajeAddVariable 14\n"
 162     "%  Time    date\n"
 163     "%  EntityType      string\n"
 164     "%  Container       string\n"
 165     "%  Value   double\n"
 166     "%EndEventDef\n"
 167     "%EventDef  PajeSubVariable 15\n"
 168     "%  Time    date\n"
 169     "%  EntityType      string\n"
 170     "%  Container       string\n"
 171     "%  Value   double\n"
 172     "%EndEventDef\n"
 173     "%EventDef  PajeStartLink   16\n"
 174     "%  Time    date\n"
 175     "%  EntityType      string\n"
 176     "%  Container       string\n"
 177     "%  Value   string\n"
 178     "%  SourceContainer string\n"
 179     "%  Key     string\n"
 180     "%EndEventDef\n"
 181     "%EventDef  PajeEndLink     17\n"
 182     "%  Time    date\n"
 183     "%  EntityType      string\n"
 184     "%  Container       string\n"
 185     "%  Value   string\n"
 186     "%  DestContainer   string\n"
 187     "%  Key     string\n"
 188     "%EndEventDef\n";
 189
 190   const char *type_definitions = "1     Sim_t   0       Simulation_t\n"
 191     "1  H_t     Sim_t   m_host_t\n"
 192     "1  P_t     H_t     m_process_t\n"
 193     "3  S_t     P_t     \"Process State\"\n"
 194     "6  E       S_t     Executing\n"
 195     "6  B       S_t     Blocked\n"
 196     "6  C       S_t     Communicating\n"
 197     "5  Comm    Sim_t   P_t     P_t     Communication_t\n";
 198
 199   const char *ext = ".trace";
 200   int ext_len = strlen(ext);
 201   int len;
 202   m_host_t host;
 203   m_process_t process;
 204   xbt_fifo_item_t item = NULL;
 205
 206   xbt_assert0(msg_global, "Initialize MSG first\n");
 207   xbt_assert0(!msg_global->paje_output, "Paje output already defined\n");
 208   xbt_assert0(filename, "Need a real file name\n");
 209
 210   len = strlen(filename);
 211   if((len<ext_len) || (strncmp(filename+len-ext_len,ext,ext_len))) {
 212     CRITICAL2("%s does not end by \"%s\". It may cause troubles when using Paje\n",
 213               filename,ext);
 214   }
 215
 216   msg_global->paje_output=fopen(filename,"w");
 217   xbt_assert1(msg_global->paje_output, "Failed to open %s \n",filename);
 218
 219   fprintf(msg_global->paje_output,"%s",paje_preembule);
 220   fprintf(msg_global->paje_output,"%s",type_definitions);
 221
 222   /*    Channels    */
 223   for(i=0; i<msg_global->max_channel; i++) {
 224     fprintf(msg_global->paje_output, "6 COMM_%d Comm    \"Channel %d\"\n" ,i,i);
 225   }
 226   fprintf(msg_global->paje_output,
 227           "7    0.0     CUR     Sim_t   0       \"MSG simulation\"\n");
 228
 229   /*    Hosts       */
 230   xbt_fifo_foreach(msg_global->host,item,host,m_host_t) {
 231     PAJE_HOST_NEW(host);
 232   }
 233
 234   /*    Process       */
 235   xbt_fifo_foreach(msg_global->process_list,item,process,m_process_t) {
 236     PAJE_PROCESS_NEW(process);
 237   }
 238 }
 239
 240 /** @deprecated Defines the verbosity of the simulation.
 241  *  @ingroup msg_simulation
 242  */
 243 void MSG_set_verbosity(MSG_outputmode_t mode)
 244 {
 245   CRITICAL0("MSG_set_verbosity : Deprecated function. Use the XBT logging interface.");
 246 }
 247
 248 /** \defgroup m_channel_management    Understanding channels
 249  *  \brief This section briefly describes the channel notion of MSG
 250  *  (#m_channel_t).
 251  *
 252  *  For convenience, the simulator provides the notion of channel
 253  *  that is close to the tag notion in MPI. A channel is not a
 254  *  socket. It doesn't need to be opened neither closed. It rather
 255  *  corresponds to the ports opened on the different machines.
 256  */
 257
 258
 259 /** \ingroup m_channel_management
 260  * \brief Set the number of channel in the simulation.
 261  *
 262  * This function has to be called to fix the number of channel in the
 263    simulation before creating any host. Indeed, each channel is
 264    represented by a different mailbox on each #m_host_t. This
 265    function can then be called only once. This function takes only one
 266    parameter.
 267  * \param number the number of channel in the simulation. It has to be >0
 268  */
 269 MSG_error_t MSG_set_channel_number(int number)
 270 {
 271   xbt_assert0((msg_global) && (msg_global->max_channel == 0), "Channel number already set!");
 272
 273   msg_global->max_channel = number;
 274
 275   return MSG_OK;
 276 }
 277
 278 /** \ingroup m_simulation
 279  * \brief Set the sharing policy used for the links
 280  *
 281  * This function can be called to change the sharing policy used for the links
 282    (see \ref paper_tcp). By default the store and forward mechanism is used
 283    with a parameter equal to 0.1. This function has to be called before creating
 284    any link.
 285  * \param mode the sharing policy used for the links: can be either
 286    MSG_STORE_AND_FORWARD or MSG_TCP.
 287  * \param param a parameter for the sharing policy. It has to be >0. It is
 288    currently used only for the MSG_STORE_AND_FORWARD flavor and represents the
 289    granularity of the communications (i.e. the packet size).
 290  */
 291 MSG_error_t MSG_set_sharing_policy(MSG_sharing_t mode, double param)
 292 {
 293   CRITICAL0("MSG_set_sharing_policy: this function is now deprecated and useless. Store and forward does not exist anymore. Please stop using it.");
 294
 295   return MSG_OK;
 296 }
 297
 298 /** \ingroup m_channel_management
 299  * \brief Return the number of channel in the simulation.
 300  *
 301  * This function has to be called once the number of channel is fixed. I can't
 302    figure out a reason why anyone would like to call this function but nevermind.
 303  * \return the number of channel in the simulation.
 304  */
 305 int MSG_get_channel_number(void)
 306 {
 307   xbt_assert0((msg_global)&&(msg_global->max_channel != 0), "Channel number not set yet!");
 308
 309   return msg_global->max_channel;
 310 }
 311
 312 /** \ingroup msg_simulation
 313  * \brief Launch the MSG simulation
 314  */
 315 MSG_error_t MSG_main(void)
 316 {
 317   m_process_t process = NULL;
 318   int nbprocess,i;
 319   double elapsed_time = 0.0;
 320   int state_modifications = 1;
 321   /* Clean IO before the run */
 322   fflush(stdout);
 323   fflush(stderr);
 324
 325   surf_solve(); /* Takes traces into account. Returns 0.0 */
 326 /* xbt_fifo_size(msg_global->process_to_run) */
 327   while (1) {
 328     xbt_context_empty_trash();
 329     if(xbt_fifo_size(msg_global->process_to_run) && (elapsed_time>0)) {
 330       DEBUG0("**************************************************");
 331     }
 332     if((__stop_at_time>0) && (MSG_getClock() >= __stop_at_time)) {
 333       DEBUG0("Let's stop here!");
 334     }
 335
 336     while ((process = xbt_fifo_pop(msg_global->process_to_run))) {
 337       DEBUG3("Scheduling  %s(%d) on %s",
 338              process->name,process->simdata->PID,
 339              process->simdata->host->name);
 340       msg_global->current_process = process;
 341 /*       fflush(NULL); */
 342       xbt_context_schedule(process->simdata->context);
 343       msg_global->current_process = NULL;
 344     }
 345
 346     {
 347       surf_action_t action = NULL;
 348       surf_resource_t resource = NULL;
 349       m_task_t task = NULL;
 350
 351       void *fun = NULL;
 352       void *arg = NULL;
 353
 354       xbt_dynar_foreach(resource_list, i, resource) {
 355         if(xbt_swag_size(resource->common_public->states.failed_action_set) ||
 356            xbt_swag_size(resource->common_public->states.done_action_set))
 357           state_modifications = 1;
 358       }
 359
 360       if(!state_modifications) {
 361         DEBUG1("%g : Calling surf_solve",MSG_getClock());
 362         elapsed_time = surf_solve();
 363         DEBUG1("Elapsed_time %g",elapsed_time);
 364
 365         if (elapsed_time<0.0) {
 366           /*       fprintf(stderr, "We're done %g\n",elapsed_time); */
 367           break;
 368         }
 369       }
 370
 371       while (surf_timer_resource->extension_public->get(&fun,(void*)&arg)) {
 372         DEBUG2("got %p %p", fun, arg);
 373         if(fun==MSG_process_create_with_arguments) {
 374           process_arg_t args = arg;
 375           DEBUG2("Launching %s on %s", args->name, args->host->name);
 376           process = MSG_process_create_with_arguments(args->name, args->code,
 377                                                       args->data, args->host,
 378                                                       args->argc,args->argv);
 379           if(args->kill_time > MSG_getClock()) {
 380             surf_timer_resource->extension_public->set(args->kill_time,
 381                                                        (void*) &MSG_process_kill,
 382                                                        (void*) process);
 383           }
 384           xbt_free(args);
 385         }
 386         if(fun==MSG_process_kill) {
 387           process = arg;
 388           DEBUG3("Killing %s(%d) on %s", process->name, process->simdata->PID,
 389                  process->simdata->host->name);
 390           MSG_process_kill(process);
 391         }
 392       }
 393
 394       xbt_dynar_foreach(resource_list, i, resource) {
 395         while ((action =
 396                 xbt_swag_extract(resource->common_public->states.
 397                                  failed_action_set))) {
 398           task = action->data;
 399           if(task) {
 400             int _cursor;
 401             DEBUG1("** %s failed **",task->name);
 402             xbt_dynar_foreach(task->simdata->sleeping,_cursor,process) {
 403               DEBUG3("\t preparing to wake up %s(%d) on %s",
 404                      process->name,process->simdata->PID,
 405                      process->simdata->host->name);
 406               xbt_fifo_unshift(msg_global->process_to_run, process);
 407             }
 408             process=NULL;
 409           }
 410         }
 411         while ((action =
 412                 xbt_swag_extract(resource->common_public->states.
 413                                  done_action_set))) {
 414           task = action->data;
 415           if(task) {
 416             int _cursor;
 417             DEBUG1("** %s done **",task->name);
 418             xbt_dynar_foreach(task->simdata->sleeping,_cursor,process) {
 419               DEBUG3("\t preparing to wake up %s(%d) on %s",
 420                      process->name,process->simdata->PID,
 421                      process->simdata->host->name);
 422               xbt_fifo_unshift(msg_global->process_to_run, process);
 423             }
 424             process=NULL;
 425           }
 426         }
 427       }
 428     }
 429     state_modifications = 0;
 430   }
 431
 432   if ((nbprocess=xbt_fifo_size(msg_global->process_list)) == 0) {
 433     fprintf(stderr,
 434             "MSG: Congratulations ! Simulation terminated : all process are over\n");
 435     return MSG_OK;
 436   } else {
 437     xbt_fifo_item_t item = NULL;
 438     fprintf(stderr,"MSG: Oops ! Deadlock or code not perfectly clean.\n");
 439     fprintf(stderr,"MSG: %d processes are still running, waiting for something.\n",
 440             nbprocess);
 441     /*  List the process and their state */
 442     fprintf(stderr,"MSG: <process>(<pid>) on <host>: <status>.\n");
 443     xbt_fifo_foreach(msg_global->process_list,item,process,m_process_t) {
 444       simdata_process_t p_simdata = (simdata_process_t) process->simdata;
 445       simdata_host_t h_simdata=(simdata_host_t)p_simdata->host->simdata;
 446
 447
 448       fprintf(stderr,"MSG:  %s(%d) on %s: ",
 449              process->name,p_simdata->PID,
 450              p_simdata->host->name);
 451
 452       if (process->simdata->blocked)
 453         fprintf(stderr,"[blocked] ");
 454       if (process->simdata->suspended)
 455         fprintf(stderr,"[suspended] ");
 456
 457       for (i=0; i<msg_global->max_channel; i++) {
 458         if (h_simdata->sleeping[i] == process) {
 459           fprintf(stderr,"Listening on channel %d.\n",i);
 460           break;
 461         }
 462       }
 463       if (i==msg_global->max_channel) {
 464         if(p_simdata->waiting_task) {
 465           if(p_simdata->waiting_task->simdata->compute) {
 466             if(p_simdata->put_host)
 467               fprintf(stderr,"Trying to send a task to Host %s, channel %d.\n",
 468                       p_simdata->put_host->name, p_simdata->put_channel);
 469             else
 470               fprintf(stderr,"Waiting for %s to finish.\n",p_simdata->waiting_task->name);
 471           } else if (p_simdata->waiting_task->simdata->comm)
 472             fprintf(stderr,"Waiting for %s to be finished transfered.\n",
 473                     p_simdata->waiting_task->name);
 474           else
 475             fprintf(stderr,"UNKNOWN STATUS. Please report this bug.\n");
 476         }
 477         else { /* Must be trying to put a task somewhere */
 478           fprintf(stderr,"UNKNOWN STATUS. Please report this bug.\n");
 479         }
 480       }
 481     }
 482     if(XBT_LOG_ISENABLED(msg, xbt_log_priority_debug) ||
 483        XBT_LOG_ISENABLED(global, xbt_log_priority_debug)) {
 484       DEBUG0("Aborting!");
 485       xbt_abort();
 486     }
 487
 488     return MSG_WARNING;
 489   }
 490 }
 491
 492 /** \ingroup msg_simulation
 493  * \brief Kill all running process
 494
 495  * \param reset_PIDs should we reset the PID numbers. A negative
 496  *   number means no reset and a positive number will be used to set the PID
 497  *   of the next newly created process.
 498  */
 499 int MSG_process_killall(int reset_PIDs)
 500 {
 501   m_process_t p = NULL;
 502   m_process_t self = MSG_process_self();
 503
 504   while((p=xbt_fifo_shift(msg_global->process_list))) {
 505     if(p!=self) MSG_process_kill(p);
 506   }
 507
 508   if(reset_PIDs>0) {
 509     msg_global->PID = reset_PIDs;
 510     msg_global->session++;
 511  }
 512
 513   xbt_context_empty_trash();
 514
 515   if(self) {
 516     xbt_context_yield();
 517   }
 518
 519   return msg_global->PID;
 520 }
 521
 522 /** \ingroup msg_simulation
 523  * \brief Clean the MSG simulation
 524  */
 525 MSG_error_t MSG_clean(void)
 526 {
 527   xbt_fifo_item_t i = NULL;
 528   m_host_t h = NULL;
 529   m_process_t p = NULL;
 530
 531
 532   while((p=xbt_fifo_shift(msg_global->process_list))) {
 533     MSG_process_kill(p);
 534   }
 535   xbt_context_exit();
 536
 537   xbt_fifo_foreach(msg_global->host,i,h,m_host_t) {
 538     __MSG_host_destroy(h);
 539   }
 540   xbt_fifo_free(msg_global->host);
 541   xbt_fifo_free(msg_global->process_to_run);
 542   xbt_fifo_free(msg_global->process_list);
 543   xbt_dict_free(&(msg_global->registered_functions));
 544
 545   if(msg_global->paje_output) {
 546     fclose(msg_global->paje_output);
 547     msg_global->paje_output = NULL;
 548   }
 549   free(msg_global);
 550   surf_finalize();
 551
 552   return MSG_OK;
 553 }
 554
 555
 556 /** \ingroup msg_easier_life
 557  * \brief A clock (in second).
 558  * \deprecated Use MSG_get_clock
 559  */
 560 double MSG_getClock(void) {
 561   return surf_get_clock();
 562 }
 563
 564 /** \ingroup msg_easier_life
 565  * \brief A clock (in second).
 566  */
 567 double MSG_get_clock(void) {
 568   return surf_get_clock();
 569 }
 570