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