src/msg/global.c

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