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